Home

SourceFlow.Net - Complete Guide

Introduction

SourceFlow.Net is a modern, lightweight, and extensible .NET framework designed for building scalable event-sourced applications using Domain-Driven Design (DDD) principles and Command Query Responsibility Segregation (CQRS) patterns. Built for .NET 8+ with performance and developer experience as core priorities.

What Makes SourceFlow.Net Special?

SourceFlow.Net provides a complete toolkit for event sourcing, domain modeling, and command/query separation, enabling developers to build maintainable, scalable applications with a strong foundation in proven architectural patterns.

Key Features

🏗️ Domain-Driven Design Support - First-class support for aggregates, entities, value objects
⚡ CQRS Implementation - Complete command/query separation with optimized read models
📊 Event Sourcing Foundation - Event-first design with full audit trail
🧱 Clean Architecture - Clear separation of concerns and dependency management
💾 Flexible Persistence - Multiple storage options including Entity Framework Core
☁️ Cloud-Native Messaging - AWS SQS/SNS integration for distributed command and event processing
🔐 Message Security - KMS envelope encryption and sensitive data masking for cloud messages
🔄 Event Replay - Built-in command replay for debugging and state reconstruction
🎯 Type Safety - Strongly-typed commands, events, and projections
📦 Dependency Injection - Seamless integration with .NET DI container
📈 OpenTelemetry Integration - Built-in distributed tracing and metrics for operations at scale
⚡ Memory Optimization - ArrayPool-based optimization for extreme throughput scenarios
🛡️ Resilience Patterns - Polly integration for fault tolerance with retry policies and circuit breakers

Core Concepts

Event Sourcing

Event Sourcing is an architectural pattern where the state of an application is determined by a sequence of events. Instead of storing the current state directly, the system stores all the events that have occurred, allowing for complete state reconstruction at any point in time.

Key Benefits:

Complete Audit Trail: Every change is recorded as an immutable event
Time Travel: Reconstruct system state at any point in history
Debugging: Full visibility into how the system reached its current state
Scalability: Events can be replayed to build multiple read models

Example in SourceFlow.Net:

// Events are immutable records of what happenedpublicclassAccountCreated:Event<BankAccount>{publicAccountCreated(BankAccountpayload):base(payload){}}publicclassMoneyDeposited:Event<BankAccount>{publicMoneyDeposited(BankAccountpayload):base(payload){}}

Domain-Driven Design (DDD)

Domain-Driven Design is a software design approach that focuses on modeling software to match the business domain. It emphasizes collaboration between technical and domain experts to create a shared understanding of the problem space.

Core DDD Elements in SourceFlow.Net:

Entities: Objects with unique identity

publicclassBankAccount:IEntity{publicintId{get;set;}publicstringAccountName{get;set;}publicdecimalBalance{get;set;}publicboolIsClosed{get;set;}publicDateTimeCreatedOn{get;set;}}

Aggregates: Coordinate business logic and ensure consistency

publicclassAccountAggregate:Aggregate<BankAccount>{publicvoidCreateAccount(intaccountId,stringholder,decimalamount){// Business logic validationSend(newCreateAccount(newCreateAccountPayload{Id=accountId,AccountName=holder,InitialAmount=amount}));}}

Sagas: Orchestrate long-running business processes

publicclassAccountSaga:Saga<BankAccount>,IHandles<CreateAccount>{publicasyncTaskHandle(CreateAccountcommand){// Validate, persist, and raise eventsvaraccount=newBankAccount{/* ... */};awaitrepository.Persist(account);awaitRaise(newAccountCreated(account));}}

Command Query Responsibility Segregation (CQRS)

CQRS separates read and write operations, allowing for optimized data models for different purposes.

Commands: Represent intent to change state

publicclassCreateAccount:Command<CreateAccountPayload>{// Parameterless constructor required for deserializationpublicCreateAccount():base(){}publicCreateAccount(CreateAccountPayloadpayload):base(payload){}}

Queries: Handled through optimized view models

publicclassAccountViewModel:IViewModel{publicintId{get;set;}publicstringAccountName{get;set;}publicdecimalCurrentBalance{get;set;}publicDateTimeLastUpdated{get;set;}publicintTransactionCount{get;set;}}

Projections: Update read models based on events

publicclassAccountProjection:IProjectOn<AccountCreated>,IProjectOn<MoneyDeposited>{publicasyncTaskApply(AccountCreated@event){varview=newAccountViewModel{Id=@event.Payload.Id,AccountName=@event.Payload.AccountName,CurrentBalance=@event.Payload.Balance};awaitprovider.Push(view);}}

Architecture Overview

High-Level Architecture

Component Interactions

Aggregates encapsulate business logic and send commands
Command Bus routes commands to appropriate saga handlers
Sagas handle commands and maintain consistency across aggregates
Sagas persist entities to the Entity Store
Sagas raise events to the Event Queue
Event Queue dispatches events to subscribers
Views are projections that update read models (ViewModels) based on events
Command Store persists commands for replay capability
Entity Store persists root aggregates (entities) within bounded context
ViewModel Store persists transformed view models from events

Entity Framework Stores provide persistence using EF Core with support for multiple databases

Cloud-Native Extension

When the SourceFlow.Cloud.AWS package is added, the architecture extends to distributed systems:

Cloud Command Dispatcher routes commands to Amazon SQS queues (Standard or FIFO)
Cloud Event Dispatcher publishes events to Amazon SNS topics with fan-out to subscribers
Bus Bootstrapper (IHostedService) auto-provisions queues, topics, and subscriptions at startup
Cloud Listeners poll SQS queues and feed messages back into the local Command Bus / Event Queue
Idempotency Service prevents duplicate message processing across multiple instances
KMS Encryption provides envelope encryption for sensitive message payloads

┌─────────────────────────────────────────────────────────────┐
│ Application Instance │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │
│ │ Aggregate │───>│ Command │───>│ SQS Command │ │
│ │ │ │ Bus │ │ Dispatcher │ │
│ └──────────┘ └──────────┘ └──────────┬───────────┘ │
│ │ │
│ ┌──────────┐ ┌──────────┐ ┌──────────▼───────────┐ │
│ │ Views │<───│ Event │<───│ SNS Event │ │
│ │ │ │ Queue │ │ Dispatcher │ │
│ └──────────┘ └──────────┘ └──────────────────────┘ │
│ │
│ ┌──────────────────────┐ ┌──────────────────────────┐ │
│ │ SQS Command Listener │ │ Idempotency Service │ │
│ │ (polls queues) │ │ (duplicate detection) │ │
│ └──────────────────────┘ └──────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
│ │
▼ ▼
┌───────────┐ ┌────────────┐
│ Amazon │ │ Amazon │
│ SQS │ │ SNS │
│ Queues │<────────────────│ Topics │
└───────────┘ (subscription) └────────────┘

Getting Started

Installation

# Install the core package
dotnet add package SourceFlow
# Install Entity Framework persistence
dotnet add package SourceFlow.Stores.EntityFramework
# Install AWS cloud messaging (optional)
dotnet add package SourceFlow.Cloud.AWS

Basic Setup

// Program.csusingSourceFlow;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;usingMicrosoft.Extensions.DependencyInjection;usingMicrosoft.Extensions.Logging;varservices=newServiceCollection();// Add loggingservices.AddLogging(builder =>{builder.AddConsole();builder.SetMinimumLevel(LogLevel.Information);});// Register entity and view model types BEFORE building service providerEntityDbContext.RegisterAssembly(typeof(Program).Assembly);ViewModelDbContext.RegisterAssembly(typeof(Program).Assembly);// Configure SourceFlow with automatic discoveryservices.UseSourceFlow(typeof(Program).Assembly);// Add Entity Framework stores with SQL Server (default)services.AddSourceFlowEfStores("Server=localhost;Database=SourceFlow;Integrated Security=true;TrustServerCertificate=true;");varserviceProvider=services.BuildServiceProvider();// Initialize databasesvarcommandContext=serviceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.EnsureCreatedAsync();varentityContext=serviceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();varviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.EnsureCreatedAsync();viewModelContext.ApplyMigrations();// Start using SourceFlowvaraggregateFactory=serviceProvider.GetRequiredService<IAggregateFactory>();varaccountAggregate=awaitaggregateFactory.Create<IAccountAggregate>();accountAggregate.CreateAccount(1,"John Doe",1000m);

For other database providers (PostgreSQL, MySQL, SQLite), see EntityFramework Usage Examples.

Framework Components

1. Aggregates

Aggregates are the primary building blocks that encapsulate business logic and coordinate with the command bus.

publicabstractclassAggregate<TEntity>:IAggregatewhereTEntity:class,IEntity{protectedICommandPublishercommandPublisher;protectedILoggerlogger;// Send commands to command busprotectedasyncTaskSend(ICommandcommand);// Subscribe to external eventspublicvirtualTaskOn(IEvent@event);}

Key Features:

Command publishing
Event subscription for external changes
Logger integration
Generic entity support

2. Sagas

Sagas handle commands and coordinate business processes, maintaining consistency across aggregate boundaries.

publicabstractclassSaga<TEntity>:ISagawhereTEntity:class,IEntity{protectedIEntityStoreAdapterrepository;protectedICommandPublishercommandPublisher;protectedIEventQueueeventQueue;protectedILoggerlogger;// Publish commandsprotectedasyncTaskPublish<TCommand>(TCommandcommand);// Raise eventsprotectedasyncTaskRaise<TEvent>(TEvent@event);}

Key Features:

Dynamic command handling via IHandles<TCommand>
Event publishing
Repository access for persistence
Built-in logging

3. Command Bus

The command bus routes commands to appropriate saga handlers and manages command persistence.

publicinterfaceICommandBus{// Publish commands to sagasTaskPublish<TCommand>(TCommandcommand)whereTCommand:ICommand;// Event dispatchers for command lifecycleeventEventHandler<ICommand>Dispatchers;}

4. Event Queue

The event queue manages event flow and dispatches events to subscribers.

publicinterfaceIEventQueue{// Enqueue events for processingTaskEnqueue<TEvent>(TEvent@event)whereTEvent:IEvent;// Event dispatcherseventEventHandler<IEvent>Dispatchers;}

5. Stores (Persistence Layer)

SourceFlow.Net defines three core store interfaces:

ICommandStore

Persists commands for event sourcing and replay

publicinterfaceICommandStore{TaskSave(ICommandcommand);Task<IEnumerable<ICommand>>Load(intentityId);}

IEntityStore

Persists domain entities

publicinterfaceIEntityStore{TaskPersist<TEntity>(TEntityentity)whereTEntity:class,IEntity;Task<TEntity>Get<TEntity>(intid)whereTEntity:class,IEntity;TaskDelete<TEntity>(TEntityentity)whereTEntity:class,IEntity;}

IViewModelStore

Persists read models (projections)

publicinterfaceIViewModelStore{TaskPersist<TViewModel>(TViewModelmodel)whereTViewModel:class,IViewModel;Task<TViewModel>Get<TViewModel>(intid)whereTViewModel:class,IViewModel;TaskDelete<TViewModel>(TViewModelmodel)whereTViewModel:class,IViewModel;}

Persistence with Entity Framework

SourceFlow.Stores.EntityFramework provides production-ready persistence using Entity Framework Core with support for multiple database providers.

Features

✅ Multiple Database Support: SQL Server, PostgreSQL, SQLite, and more
✅ Flexible Configuration: Single or separate connection strings per store
✅ Dynamic Type Registration: Runtime registration of entities and view models
✅ Migration Support: Manual table creation bypassing EF Core model caching
✅ Thread-Safe: Designed for concurrent access
✅ Optimized Tracking: Proper EF Core change tracking management
✅ Production-Ready Enhancements: Resilience, observability, and memory optimization

Installation

dotnet add package SourceFlow.Stores.EntityFramework

Configuration Options

1. Single Connection String (All Stores)

Use the same database for all stores:

services.AddSourceFlowEfStores("Server=localhost;Database=SourceFlow;Integrated Security=true;");

2. Separate Connection Strings

Use different databases for each store:

services.AddSourceFlowEfStores(commandConnectionString:"Server=localhost;Database=SourceFlow_Commands;...",entityConnectionString:"Server=localhost;Database=SourceFlow_Entities;...",viewModelConnectionString:"Server=localhost;Database=SourceFlow_Views;...");

3. Configuration-Based

Read from appsettings.json:

{
"ConnectionStrings": {
"SourceFlow.Default": "Server=localhost;Database=SourceFlow;Integrated Security=true;",
"SourceFlow.Command": "Server=localhost;Database=Commands;...",
"SourceFlow.Entity": "Server=localhost;Database=Entities;...",
"SourceFlow.ViewModel": "Server=localhost;Database=Views;..."
}
}

services.AddSourceFlowEfStores(configuration);

4. Options Pattern

Configure using options:

services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString="Server=localhost;Database=SourceFlow;...";// Or specify individual connection stringsoptions.CommandConnectionString="...";options.EntityConnectionString="...";options.ViewModelConnectionString="...";});

5. Custom Database Provider

Use PostgreSQL, SQLite, or other providers:

// PostgreSQLservices.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseNpgsql("Host=localhost;Database=sourceflow;Username=postgres;Password=..."));// SQLiteservices.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseSqlite("Data Source=sourceflow.db"));// In-Memory (for testing)services.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseInMemoryDatabase("SourceFlowTest"));

6. Different Providers Per Store

Use different database types for each store:

services.AddSourceFlowEfStoresWithCustomProviders(commandContextConfig: options =>options.UseSqlServer("..."),entityContextConfig: options =>options.UseNpgsql("..."),viewModelContextConfig: options =>options.UseSqlite("..."));

Dynamic Type Registration

Entity Framework requires types to be registered before creating the database schema. SourceFlow.Stores.EntityFramework provides multiple registration strategies:

1. Explicit Type Registration

Register specific types before database initialization:

// In your startup or test setupEntityDbContext.RegisterEntityType<BankAccount>();EntityDbContext.RegisterEntityType<Customer>();ViewModelDbContext.RegisterViewModelType<AccountViewModel>();ViewModelDbContext.RegisterViewModelType<CustomerViewModel>();// Then build service provider and ensure databases are createdvarserviceProvider=services.BuildServiceProvider();varentityContext=serviceProvider.GetRequiredService<EntityDbContext>();entityContext.Database.EnsureCreated();entityContext.ApplyMigrations();// Creates tables for registered typesvarviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();viewModelContext.Database.EnsureCreated();viewModelContext.ApplyMigrations();// Creates tables for registered view models

2. Assembly Scanning

Register all types from an assembly:

// Register the test or application assemblyEntityDbContext.RegisterAssembly(typeof(BankAccount).Assembly);ViewModelDbContext.RegisterAssembly(typeof(AccountViewModel).Assembly);varserviceProvider=services.BuildServiceProvider();// Apply migrations to create tablesvarentityContext=serviceProvider.GetRequiredService<EntityDbContext>();entityContext.Database.EnsureCreated();entityContext.ApplyMigrations();varviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();viewModelContext.Database.EnsureCreated();viewModelContext.ApplyMigrations();

3. Auto-Discovery (Fallback)

The DbContexts automatically discover types from loaded assemblies (fallback mechanism):

// Just ensure databases are createdvarentityContext=serviceProvider.GetRequiredService<EntityDbContext>();entityContext.Database.EnsureCreated();varviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();viewModelContext.Database.EnsureCreated();// Note: This may not catch all types reliably; explicit registration is recommended

Table Naming Convention

All dynamically created tables use the T prefix:

BankAccount entity → TBankAccount table
AccountViewModel → TAccountViewModel table
Customer entity → TCustomer table

This convention helps distinguish dynamically created tables from EF Core's built-in tables.

Migration Helper

The DbContextMigrationHelper manually creates database schemas, bypassing EF Core's model caching:

// Called automatically by ApplyMigrations()publicstaticvoidCreateEntityTables(EntityDbContextcontext,IEnumerable<Type>entityTypes){// Creates tables with proper columns and primary keys// Supports int, long, string, bool, DateTime, decimal, double, float, byte[], enums}publicstaticvoidCreateViewModelTables(ViewModelDbContextcontext,IEnumerable<Type>viewModelTypes){// Creates tables for view models}

Complete Setup Example

usingSourceFlow;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;usingMicrosoft.Extensions.DependencyInjection;usingMicrosoft.Extensions.Logging;usingMicrosoft.EntityFrameworkCore;varservices=newServiceCollection();// Add loggingservices.AddLogging(builder =>{builder.AddConsole();builder.SetMinimumLevel(LogLevel.Information);});// Register types BEFORE building service providerEntityDbContext.RegisterAssembly(typeof(BankAccount).Assembly);ViewModelDbContext.RegisterAssembly(typeof(AccountViewModel).Assembly);// Configure SourceFlowservices.UseSourceFlow(typeof(Program).Assembly);// Add Entity Framework stores (SQL Server by default)services.AddSourceFlowEfStores("Server=localhost;Database=SourceFlow;Integrated Security=true;TrustServerCertificate=true;");// Or use custom provider for other databases:// services.AddSourceFlowEfStoresWithCustomProvider(options =>// options.UseNpgsql("Host=localhost;Database=sourceflow;Username=postgres;Password=..."));varserviceProvider=services.BuildServiceProvider();// Ensure all databases are created and migratedvarcommandContext=serviceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.EnsureCreatedAsync();varentityContext=serviceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();// Create tables for registered entity typesvarviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.EnsureCreatedAsync();viewModelContext.ApplyMigrations();// Create tables for registered view model types// Start using SourceFlowvaraggregateFactory=serviceProvider.GetRequiredService<IAggregateFactory>();varaccountAggregate=awaitaggregateFactory.Create<IAccountAggregate>();accountAggregate.CreateAccount(1,"John Doe",1000m);

Testing with In-Memory Database

For unit and integration tests, use SQLite in-memory databases with proper setup:

usingNUnit.Framework;usingMicrosoft.Data.Sqlite;usingMicrosoft.EntityFrameworkCore;usingMicrosoft.Extensions.DependencyInjection;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;usingSourceFlow.Stores.EntityFramework.Options;usingSourceFlow.Stores.EntityFramework.Services;usingSourceFlow.Stores.EntityFramework.Stores;[TestFixture]publicclassBankAccountIntegrationTests{privateServiceProvider?_serviceProvider;privateSqliteConnection?_connection;[SetUp]publicvoidSetup(){// Clear previous registrationsEntityDbContext.ClearRegistrations();ViewModelDbContext.ClearRegistrations();// Register test typesEntityDbContext.RegisterEntityType<BankAccount>();ViewModelDbContext.RegisterViewModelType<AccountViewModel>();// Create shared in-memory SQLite connection for all contexts_connection=newSqliteConnection("DataSource=:memory:");_connection.Open();varservices=newServiceCollection();// Add logging for better test diagnosticsservices.AddLogging(builder =>{builder.AddConsole();builder.SetMinimumLevel(LogLevel.Debug);});// Configure SQLite with shared connection// Use EnableServiceProviderCaching(false) to avoid EF Core 9.0 multiple provider conflictsservices.AddDbContext<CommandDbContext>(options =>options.UseSqlite(_connection).EnableServiceProviderCaching(false));services.AddDbContext<EntityDbContext>(options =>options.UseSqlite(_connection).EnableServiceProviderCaching(false));services.AddDbContext<ViewModelDbContext>(options =>options.UseSqlite(_connection).EnableServiceProviderCaching(false));// Register SourceFlowEfOptions with default settingsvarefOptions=newSourceFlowEfOptions();services.AddSingleton(efOptions);// Register common services manually (avoids provider conflicts)services.AddScoped<IDatabaseResiliencePolicy,DatabaseResiliencePolicy>();services.AddScoped<IDatabaseTelemetryService,DatabaseTelemetryService>();services.AddScoped<ICommandStore,EfCommandStore>();services.AddScoped<IEntityStore,EfEntityStore>();services.AddScoped<IViewModelStore,EfViewModelStore>();// Register SourceFlowservices.UseSourceFlow(Assembly.GetExecutingAssembly());_serviceProvider=services.BuildServiceProvider();// Create all database schemasvarcommandContext=_serviceProvider.GetRequiredService<CommandDbContext>();commandContext.Database.EnsureCreated();varentityContext=_serviceProvider.GetRequiredService<EntityDbContext>();entityContext.Database.EnsureCreated();entityContext.ApplyMigrations();// Create tables for registered entity typesvarviewModelContext=_serviceProvider.GetRequiredService<ViewModelDbContext>();viewModelContext.Database.EnsureCreated();viewModelContext.ApplyMigrations();// Create tables for registered view model types}[TearDown]publicvoidTearDown(){// Clean up resources_connection?.Close();_connection?.Dispose();_serviceProvider?.Dispose();}[Test]publicasyncTaskCreateAccount_StoresInDatabase(){// ArrangevaraggregateFactory=_serviceProvider.GetRequiredService<IAggregateFactory>();varaccountAggregate=awaitaggregateFactory.Create<IAccountAggregate>();// ActaccountAggregate.CreateAccount(1,"John Doe",1000m);// Wait for async processingawaitTask.Delay(100);// AssertvarentityStore=_serviceProvider.GetRequiredService<IEntityStoreAdapter>();varaccount=awaitentityStore.Get<BankAccount>(1);Assert.That(account,Is.Not.Null);Assert.That(account.AccountName,Is.EqualTo("John Doe"));Assert.That(account.Balance,Is.EqualTo(1000m));}}

EntityFramework Usage Examples

This section provides practical examples for common scenarios using SourceFlow.Stores.EntityFramework.

Example 1: Simple Console Application with SQL Server

Complete working example for a console application:

usingSystem;usingSystem.Threading.Tasks;usingMicrosoft.Extensions.DependencyInjection;usingMicrosoft.Extensions.Logging;usingSourceFlow;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;classProgram{staticasyncTaskMain(string[]args){// Setup service collectionvarservices=newServiceCollection();// Add loggingservices.AddLogging(builder =>{builder.AddConsole();builder.SetMinimumLevel(LogLevel.Information);});// Register entity and view model types BEFORE building service providerEntityDbContext.RegisterEntityType<BankAccount>();ViewModelDbContext.RegisterViewModelType<AccountViewModel>();// Configure SourceFlowservices.UseSourceFlow(typeof(Program).Assembly);// Add Entity Framework stores with SQL Serverservices.AddSourceFlowEfStores("Server=localhost;Database=SourceFlowDemo;Integrated Security=true;TrustServerCertificate=true;");varserviceProvider=services.BuildServiceProvider();// Ensure databases are createdvarcommandContext=serviceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.EnsureCreatedAsync();varentityContext=serviceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();varviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.EnsureCreatedAsync();viewModelContext.ApplyMigrations();// Use the aggregatevaraggregateFactory=serviceProvider.GetRequiredService<IAggregateFactory>();varaccountAggregate=awaitaggregateFactory.Create<IAccountAggregate>();// Execute business operationsaccountAggregate.CreateAccount(1,"Alice Smith",5000m);accountAggregate.Deposit(1,1500m);accountAggregate.Withdraw(1,500m);// Give async processing time to completeawaitTask.Delay(500);// Query the read modelvarviewModelStore=serviceProvider.GetRequiredService<IViewModelStoreAdapter>();varaccountView=awaitviewModelStore.Find<AccountViewModel>(1);Console.WriteLine($"Account: {accountView.AccountName}");Console.WriteLine($"Balance: {accountView.CurrentBalance:C}");Console.WriteLine($"Transactions: {accountView.TransactionCount}");Console.WriteLine($"Created: {accountView.CreatedDate:yyyy-MM-dd}");}}

Example 2: ASP.NET Core Web API with PostgreSQL

Complete setup for a web API using PostgreSQL:

// Program.csusingMicrosoft.EntityFrameworkCore;usingSourceFlow;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;varbuilder=WebApplication.CreateBuilder(args);// Add services to the containerbuilder.Services.AddControllers();builder.Services.AddEndpointsApiExplorer();builder.Services.AddSwaggerGen();// Register entity and view model typesEntityDbContext.RegisterAssembly(typeof(Program).Assembly);ViewModelDbContext.RegisterAssembly(typeof(Program).Assembly);// Configure SourceFlow with PostgreSQLbuilder.Services.UseSourceFlow(typeof(Program).Assembly);builder.Services.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseNpgsql(builder.Configuration.GetConnectionString("SourceFlow")));varapp=builder.Build();// Initialize databases on startupusing(varscope=app.Services.CreateScope()){varcommandContext=scope.ServiceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.EnsureCreatedAsync();varentityContext=scope.ServiceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();varviewModelContext=scope.ServiceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.EnsureCreatedAsync();viewModelContext.ApplyMigrations();}if(app.Environment.IsDevelopment()){app.UseSwagger();app.UseSwaggerUI();}app.UseHttpsRedirection();app.UseAuthorization();app.MapControllers();app.Run();

// appsettings.json{"ConnectionStrings":{"SourceFlow":"Host=localhost;Database=sourceflow;Username=postgres;Password=yourpassword"},"Logging":{"LogLevel":{"Default":"Information","Microsoft.EntityFrameworkCore":"Warning"}}}

// Controllers/AccountController.csusingMicrosoft.AspNetCore.Mvc;usingSourceFlow;[ApiController][Route("api/[controller]")]publicclassAccountController:ControllerBase{privatereadonlyIAggregateFactory_aggregateFactory;privatereadonlyIViewModelStoreAdapter_viewModelStore;privatereadonlyILogger<AccountController>_logger;publicAccountController(IAggregateFactoryaggregateFactory,IViewModelStoreAdapterviewModelStore,ILogger<AccountController>logger){_aggregateFactory=aggregateFactory;_viewModelStore=viewModelStore;_logger=logger;}[HttpPost]publicasyncTask<IActionResult>CreateAccount(CreateAccountRequestrequest){varaggregate=await_aggregateFactory.Create<IAccountAggregate>();aggregate.CreateAccount(request.Id,request.AccountName,request.InitialBalance);_logger.LogInformation("Account created: {AccountId}",request.Id);returnCreatedAtAction(nameof(GetAccount),new{id=request.Id},request);}[HttpGet("{id}")]publicasyncTask<ActionResult<AccountViewModel>>GetAccount(intid){try{varaccount=await_viewModelStore.Find<AccountViewModel>(id);returnOk(account);}catch(InvalidOperationException){returnNotFound();}}[HttpPost("{id}/deposit")]publicasyncTask<IActionResult>Deposit(intid,[FromBody]TransactionRequestrequest){varaggregate=await_aggregateFactory.Create<IAccountAggregate>();aggregate.Deposit(id,request.Amount);_logger.LogInformation("Deposited {Amount} to account {AccountId}",request.Amount,id);returnNoContent();}[HttpPost("{id}/withdraw")]publicasyncTask<IActionResult>Withdraw(intid,[FromBody]TransactionRequestrequest){try{varaggregate=await_aggregateFactory.Create<IAccountAggregate>();aggregate.Withdraw(id,request.Amount);_logger.LogInformation("Withdrew {Amount} from account {AccountId}",request.Amount,id);returnNoContent();}catch(InvalidOperationExceptionex){returnBadRequest(new{error=ex.Message});}}}publicrecordCreateAccountRequest(intId,stringAccountName,decimalInitialBalance);publicrecordTransactionRequest(decimalAmount);

Example 3: Microservices with Separate Databases

Using different databases for different stores in a microservices architecture:

// Program.cs for Banking Microservicevarbuilder=WebApplication.CreateBuilder(args);// Register typesEntityDbContext.RegisterAssembly(typeof(Program).Assembly);ViewModelDbContext.RegisterAssembly(typeof(Program).Assembly);// Configure SourceFlowbuilder.Services.UseSourceFlow(typeof(Program).Assembly);// Each store uses a different database optimized for its purposebuilder.Services.AddSourceFlowEfStoresWithCustomProviders(// Commands: PostgreSQL with JSONB support for efficient command storagecommandContextConfig: opt =>opt.UseNpgsql(builder.Configuration.GetConnectionString("CommandStore")),// Entities: SQL Server with optimized indexes for transactional workloadentityContextConfig: opt =>opt.UseSqlServer(builder.Configuration.GetConnectionString("EntityStore")),// ViewModels: SQLite for fast read queries in read-heavy scenariosviewModelContextConfig: opt =>opt.UseSqlite(builder.Configuration.GetConnectionString("ViewStore")));varapp=builder.Build();// Initialize all databasesusing(varscope=app.Services.CreateScope()){varcommandContext=scope.ServiceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.MigrateAsync();varentityContext=scope.ServiceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.MigrateAsync();entityContext.ApplyMigrations();varviewModelContext=scope.ServiceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.MigrateAsync();viewModelContext.ApplyMigrations();}app.Run();

// appsettings.json
{
"ConnectionStrings": {
"CommandStore": "Host=postgres-commands.internal;Database=banking_commands;Username=app;Password=...",
"EntityStore": "Server=sqlserver-entities.internal;Database=banking_entities;User Id=app;Password=...;",
"ViewStore": "Data Source=/data/banking_views.db"
}
}

Example 4: Production Configuration with Resilience and Observability

Complete production setup with all enterprise features:

// Program.csusingSourceFlow;usingSourceFlow.Observability;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;usingOpenTelemetry;usingMicrosoft.EntityFrameworkCore;varbuilder=WebApplication.CreateBuilder(args);// Register domain typesEntityDbContext.RegisterAssembly(typeof(Program).Assembly);ViewModelDbContext.RegisterAssembly(typeof(Program).Assembly);// Configure SourceFlow with observabilitybuilder.Services.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="BankingService";options.ServiceVersion=builder.Configuration["AppVersion"]??"1.0.0";});// Configure OpenTelemetry exportersbuilder.Services.AddOpenTelemetry().AddSourceFlowOtlpExporter(builder.Configuration["Observability:OtlpEndpoint"]).AddSourceFlowResourceAttributes(("environment",builder.Environment.EnvironmentName),("deployment.region",builder.Configuration["Deployment:Region"]),("service.instance.id",Environment.MachineName)).ConfigureSourceFlowBatchProcessing(maxQueueSize:2048,maxExportBatchSize:512,scheduledDelayMilliseconds:5000);// Register SourceFlowbuilder.Services.UseSourceFlow(typeof(Program).Assembly);// Configure Entity Framework stores with resilience and observabilitybuilder.Services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=builder.Configuration.GetConnectionString("SourceFlow");// Resilience configurationoptions.Resilience.Enabled=true;options.Resilience.Retry.MaxRetryAttempts=3;options.Resilience.Retry.BaseDelayMs=1000;options.Resilience.Retry.UseExponentialBackoff=true;options.Resilience.Retry.UseJitter=true;options.Resilience.CircuitBreaker.Enabled=true;options.Resilience.CircuitBreaker.FailureThreshold=10;options.Resilience.CircuitBreaker.BreakDurationMs=60000;options.Resilience.Timeout.Enabled=true;options.Resilience.Timeout.TimeoutMs=30000;// Observability configurationoptions.Observability.Enabled=true;options.Observability.ServiceName="BankingService.EntityFramework";options.Observability.Tracing.Enabled=true;options.Observability.Tracing.TraceDatabaseOperations=true;options.Observability.Tracing.IncludeSqlInTraces=false;// Don't log SQL in productionoptions.Observability.Tracing.SamplingRatio=0.1;// Sample 10%options.Observability.Metrics.Enabled=true;options.Observability.Metrics.CollectDatabaseMetrics=true;// Table naming conventionsoptions.EntityTableNaming.Casing=TableNameCasing.SnakeCase;options.EntityTableNaming.Pluralize=true;options.ViewModelTableNaming.Casing=TableNameCasing.SnakeCase;options.ViewModelTableNaming.Suffix="_view";});// Add health checksbuilder.Services.AddHealthChecks().AddDbContextCheck<CommandDbContext>("command-store").AddDbContextCheck<EntityDbContext>("entity-store").AddDbContextCheck<ViewModelDbContext>("viewmodel-store");varapp=builder.Build();// Initialize databasesusing(varscope=app.Services.CreateScope()){varlogger=scope.ServiceProvider.GetRequiredService<ILogger<Program>>();try{varcommandContext=scope.ServiceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.EnsureCreatedAsync();logger.LogInformation("Command store initialized");varentityContext=scope.ServiceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();logger.LogInformation("Entity store initialized");varviewModelContext=scope.ServiceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.EnsureCreatedAsync();viewModelContext.ApplyMigrations();logger.LogInformation("ViewModel store initialized");}catch(Exceptionex){logger.LogError(ex,"Failed to initialize databases");throw;}}app.MapHealthChecks("/health");app.Run();

// appsettings.Production.json
{
"ConnectionStrings": {
"SourceFlow": "Server=prod-db.internal;Database=BankingService;User Id=app_user;Password=...;Max Pool Size=100;Min Pool Size=10;"
},
"Observability": {
"OtlpEndpoint": "http://otel-collector.internal:4317"
},
"Deployment": {
"Region": "us-east-1"
},
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.EntityFrameworkCore": "Warning",
"SourceFlow": "Information"
}
}
}

Example 5: Custom Table Naming with Schema Organization

Organizing tables with custom naming conventions and schemas:

services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=connectionString;// Command store: Audit schema with snake_caseoptions.CommandTableNaming.UseSchema=true;options.CommandTableNaming.SchemaName="audit";options.CommandTableNaming.Casing=TableNameCasing.SnakeCase;// Results in: audit.command_record// Entity store: Domain schema with pluralized tablesoptions.EntityTableNaming.UseSchema=true;options.EntityTableNaming.SchemaName="domain";options.EntityTableNaming.Casing=TableNameCasing.SnakeCase;options.EntityTableNaming.Pluralize=true;// BankAccount -> domain.bank_accounts// ViewModel store: Reporting schema with view suffixoptions.ViewModelTableNaming.UseSchema=true;options.ViewModelTableNaming.SchemaName="reporting";options.ViewModelTableNaming.Casing=TableNameCasing.SnakeCase;options.ViewModelTableNaming.Suffix="_view";options.ViewModelTableNaming.Pluralize=true;// AccountViewModel -> reporting.account_views});// Create schemas before initializingvarentityContext=serviceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.ExecuteSqlRawAsync("CREATE SCHEMA IF NOT EXISTS audit");awaitentityContext.Database.ExecuteSqlRawAsync("CREATE SCHEMA IF NOT EXISTS domain");awaitentityContext.Database.ExecuteSqlRawAsync("CREATE SCHEMA IF NOT EXISTS reporting");awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();

Example 6: Background Service Processing Commands

Processing commands in a background service:

publicclassCommandProcessorBackgroundService:BackgroundService{privatereadonlyIServiceProvider_serviceProvider;privatereadonlyILogger<CommandProcessorBackgroundService>_logger;publicCommandProcessorBackgroundService(IServiceProviderserviceProvider,ILogger<CommandProcessorBackgroundService>logger){_serviceProvider=serviceProvider;_logger=logger;}protectedoverrideasyncTaskExecuteAsync(CancellationTokenstoppingToken){_logger.LogInformation("Command Processor Background Service starting");while(!stoppingToken.IsCancellationRequested){try{usingvarscope=_serviceProvider.CreateScope();varcommandStore=scope.ServiceProvider.GetRequiredService<ICommandStoreAdapter>();varaggregateFactory=scope.ServiceProvider.GetRequiredService<IAggregateFactory>();// Process any pending commands (example: replay for specific entities)varentityIds=awaitGetPendingEntityIds();foreach(varentityIdinentityIds){varcommands=awaitcommandStore.Retrieve(entityId);if(commands.Any()){_logger.LogInformation("Processing {Count} commands for entity {EntityId}",commands.Count(),entityId);// Replay commands to rebuild statevaraggregate=awaitaggregateFactory.Create<IAccountAggregate>();// Process commands...}}awaitTask.Delay(TimeSpan.FromMinutes(5),stoppingToken);}catch(Exceptionex){_logger.LogError(ex,"Error in command processor");awaitTask.Delay(TimeSpan.FromMinutes(1),stoppingToken);}}_logger.LogInformation("Command Processor Background Service stopping");}privateasyncTask<List<int>>GetPendingEntityIds(){// Implementation to identify entities needing processingreturnnewList<int>();}}// Register in Program.csbuilder.Services.AddHostedService<CommandProcessorBackgroundService>();

Example 7: Multi-Tenant Setup with Database Per Tenant

Implementing multi-tenancy with separate databases:

// ITenantProvider.cspublicinterfaceITenantProvider{stringGetCurrentTenantId();stringGetConnectionString(stringtenantId);}// TenantDbContextFactory.cspublicclassTenantDbContextFactory<TContext>whereTContext:DbContext{privatereadonlyITenantProvider_tenantProvider;privatereadonlyIServiceProvider_serviceProvider;publicTenantDbContextFactory(ITenantProvidertenantProvider,IServiceProviderserviceProvider){_tenantProvider=tenantProvider;_serviceProvider=serviceProvider;}publicTContextCreateDbContext(){vartenantId=_tenantProvider.GetCurrentTenantId();varconnectionString=_tenantProvider.GetConnectionString(tenantId);varoptionsBuilder=newDbContextOptionsBuilder<TContext>();optionsBuilder.UseSqlServer(connectionString);return(TContext)Activator.CreateInstance(typeof(TContext),optionsBuilder.Options);}}// Program.csbuilder.Services.AddScoped<ITenantProvider,HttpContextTenantProvider>();// Register SourceFlow with multi-tenant supportbuilder.Services.UseSourceFlow(typeof(Program).Assembly);// Custom multi-tenant store registrationbuilder.Services.AddScoped(sp =>{vartenantProvider=sp.GetRequiredService<ITenantProvider>();vartenantId=tenantProvider.GetCurrentTenantId();varconnectionString=tenantProvider.GetConnectionString(tenantId);varoptionsBuilder=newDbContextOptionsBuilder<EntityDbContext>();optionsBuilder.UseSqlServer(connectionString);returnnewEntityDbContext(optionsBuilder.Options);});// Similar for CommandDbContext and ViewModelDbContext

Cloud Integration (AWS)

SourceFlow.Cloud.AWS extends the framework with distributed command and event processing using Amazon SQS, SNS, and KMS. It enables multiple application instances to communicate through cloud messaging while preserving the same CQRS and event-sourcing patterns used locally.

Installation

dotnet add package SourceFlow.Cloud.AWS

Prerequisites: SourceFlow >= 2.0.0, .NET Standard 2.1 / .NET 8.0+ / .NET 9.0+ / .NET 10.0+

Quick Start

usingSourceFlow.Cloud.AWS;usingAmazon;// Register SourceFlow coreservices.UseSourceFlow(typeof(Program).Assembly);// Configure AWS cloud messagingservices.UseSourceFlowAws(
options =>{options.Region=RegionEndpoint.USEast1;options.MaxConcurrentCalls=10;},
bus =>bus.Send.Command<CreateOrderCommand>(q =>q.Queue("orders.fifo")).Command<ProcessPaymentCommand>(q =>q.Queue("payments.fifo")).Raise.Event<OrderCreatedEvent>(t =>t.Topic("order-events")).Event<PaymentProcessedEvent>(t =>t.Topic("payment-events")).Listen.To.CommandQueue("orders.fifo").CommandQueue("payments.fifo").Subscribe.To.Topic("order-events").Topic("payment-events"));

This registers AWS dispatchers, configures routing, starts SQS listeners, and automatically provisions queues/topics/subscriptions at startup via the AwsBusBootstrapper hosted service.

Bus Configuration API

The fluent bus configuration API maps commands to SQS queues and events to SNS topics:

Send Commands to SQS Queues

.Send.Command<CreateOrderCommand>(q =>q.Queue("orders.fifo"))// FIFO queue.Command<SendEmailCommand>(q =>q.Queue("notifications"))// Standard queue

FIFO queues (name ends with .fifo): Exactly-once processing, strict ordering per message group, content-based deduplication
Standard queues: High throughput, at-least-once delivery, best-effort ordering

Raise Events to SNS Topics

.Raise.Event<OrderCreatedEvent>(t =>t.Topic("order-events")).Event<PaymentProcessedEvent>(t =>t.Topic("payment-events"))

Events are published to SNS topics using the publish-subscribe pattern with fan-out to all subscribed queues.

Listen to Command Queues

.Listen.To.CommandQueue("orders.fifo").CommandQueue("payments.fifo")

Listeners poll SQS queues and feed received messages back into the local Command Bus for saga processing.

Subscribe to Event Topics

.Subscribe.To.Topic("order-events").Topic("payment-events")

SNS topics are subscribed to the first configured command queue, enabling event-driven cross-service communication.

Configuration Options

Option	Type	Default	Description
`Region`	`RegionEndpoint`	Required	AWS region for SQS/SNS/KMS
`EnableCommandRouting`	`bool`	`true`	Enable command dispatching to SQS
`EnableEventRouting`	`bool`	`true`	Enable event publishing to SNS
`EnableCommandListener`	`bool`	`true`	Enable SQS command polling
`EnableEventListener`	`bool`	`true`	Enable SNS event listener
`MaxConcurrentCalls`	`int`	`10`	Max concurrent message processing
`EnableEncryption`	`bool`	`false`	Enable KMS message encryption
`KmsKeyId`	`string`	`null`	KMS key ID or alias

Message Encryption (KMS)

Enable envelope encryption for sensitive message payloads:

services.UseSourceFlowAws(
options =>{options.Region=RegionEndpoint.USEast1;options.EnableEncryption=true;options.KmsKeyId="alias/sourceflow-key";},
bus => ...);

Encryption flow: Generate data key (KMS) -> Encrypt message (data key) -> Encrypt data key (KMS master key) -> Store encrypted message + encrypted data key in SQS.

Decryption flow: Retrieve encrypted message -> Decrypt data key (KMS) -> Decrypt message (data key) -> Plaintext message.

Automatic Resource Provisioning

The AwsBusBootstrapper runs as an IHostedService at startup and automatically creates:

SQS queues: Standard and FIFO with dead letter queues, configurable retention and visibility timeout
SNS topics: With display names
SNS-to-SQS subscriptions: Raw message delivery enabled, with queue policy updates for SNS publish permissions

All provisioning operations are idempotent — safe to run on every application startup.

Idempotency

In-Memory (Single Instance)

Automatically registered when using UseSourceFlowAws(). Suitable for single-instance deployments.

SQL-Based (Multi-Instance)

For production with multiple application instances processing the same queues:

// Install: dotnet add package SourceFlow.Stores.EntityFramework// Register SQL-backed idempotency (replaces in-memory default)services.AddSourceFlowIdempotency(connectionString:configuration.GetConnectionString("IdempotencyStore"),cleanupIntervalMinutes:60);// Then configure AWSservices.UseSourceFlowAws(options => ..., bus => ...);

The EfIdempotencyService uses database transactions for thread-safe duplicate detection across instances, with automatic background cleanup of expired records.

Command and Event Flow

Command Flow (SQS):

Aggregate.Send(command)
→ CommandBus (assigns sequence number)
→ AwsSqsCommandDispatcher (checks routing, encrypts if enabled)
→ SQS Queue (message persisted)
→ AwsSqsCommandListener (polls queue, decrypts, idempotency check)
→ CommandBus.Publish (local processing)
→ Saga handles command

Event Flow (SNS → SQS):

Saga.Raise(event)
→ EventQueue (enqueues event)
→ AwsSnsEventDispatcher (checks routing, publishes to SNS)
→ SNS Topic (fan-out to subscribers)
→ SQS Queue (subscribed to topic)
→ AwsSqsCommandListener (polls queue, idempotency check)
→ EventQueue.Enqueue (local processing)
→ Views/Aggregates handle event

Health Checks

services.AddHealthChecks().AddCheck<AwsHealthCheck>("aws");

Checks SQS connectivity, SNS connectivity, KMS access (if encryption enabled), and queue/topic existence.

Observability

Activity Source: SourceFlow.Cloud.AWS

Traces:

AwsSqsCommandDispatcher.Dispatch — Command dispatch to SQS
AwsSnsEventDispatcher.Dispatch — Event publish to SNS
AwsSqsCommandListener.ProcessMessage — Message processing from SQS

Metrics:

sourceflow.aws.command.dispatched / dispatch_duration / dispatch_error
sourceflow.aws.event.published / publish_duration / publish_error
sourceflow.aws.message.received / processed / processing_duration / processing_error

Trace context is propagated via SQS message attributes for end-to-end distributed tracing.

Local Development with LocalStack

LocalStack provides local AWS service emulation for development and testing.

Quick Start (Recommended)

# PowerShell (Windows)
./tests/SourceFlow.Cloud.AWS.Tests/run-integration-tests.ps1
# Bash (Linux/macOS/WSL)
./tests/SourceFlow.Cloud.AWS.Tests/run-integration-tests.sh

The scripts start a LocalStack Docker container, wait for SQS/SNS/KMS services, set environment variables, and run integration tests. Use -KeepRunning / --keep to leave the container running.

Manual Setup

docker run -d --name sourceflow-localstack \
-p 4566:4566 \
-e SERVICES=sqs,sns,kms \
-e EAGER_SERVICE_LOADING=1 \
localstack/localstack:latest

Environment Variables

export AWS_ENDPOINT_URL=http://localhost:4566
export AWS_DEFAULT_REGION=us-east-1
# LocalStack uses dummy credentials — test fixtures use BasicAWSCredentials("test", "test")

Integration Tests

[Trait("Category","Integration")][Trait("Category","RequiresLocalStack")]publicclassMyIntegrationTests:LocalStackRequiredTestBase{[Fact]publicasyncTaskShould_Process_Command_Through_SQS(){// Test against local SQS/SNS/KMS}}

# Run integration tests (LocalStack must be running)
dotnet test --filter "Category=Integration&Category=RequiresLocalStack"

Complete Cloud Application Example

End-to-end setup with EF persistence, cloud messaging, and SQL-backed idempotency:

varbuilder=WebApplication.CreateBuilder(args);// 1. Register domain typesEntityDbContext.RegisterAssembly(typeof(Program).Assembly);ViewModelDbContext.RegisterAssembly(typeof(Program).Assembly);// 2. Register SourceFlow corebuilder.Services.UseSourceFlow(typeof(Program).Assembly);// 3. Register EF persistence storesbuilder.Services.AddSourceFlowEfStores(builder.Configuration, options =>{options.UseCommandStore("CommandStore");options.UseEntityStore("EntityStore");options.UseViewModelStore("ViewModelStore");});// 4. Register SQL-backed idempotency for multi-instance deploymentsbuilder.Services.AddSourceFlowIdempotency(connectionString:builder.Configuration.GetConnectionString("IdempotencyStore"),cleanupIntervalMinutes:60);// 5. Configure AWS cloud messagingbuilder.Services.UseSourceFlowAws(
options =>{options.Region=RegionEndpoint.USEast1;options.EnableEncryption=true;options.KmsKeyId="alias/sourceflow-key";options.MaxConcurrentCalls=10;},
bus =>bus.Send.Command<CreateOrderCommand>(q =>q.Queue("orders.fifo")).Raise.Event<OrderCreatedEvent>(t =>t.Topic("order-events")).Listen.To.CommandQueue("orders.fifo").Subscribe.To.Topic("order-events"));// 6. Health checksbuilder.Services.AddHealthChecks().AddDbContextCheck<EntityDbContext>("entity-store").AddCheck<AwsHealthCheck>("aws");// 7. Observabilitybuilder.Services.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="OrderService";});varapp=builder.Build();app.MapHealthChecks("/health");app.Run();

IAM Permissions

Development (broad access):

{
"Statement": [
{ "Action": ["sqs:*"], "Resource": "arn:aws:sqs:*:*:*", "Effect": "Allow" },
{ "Action": ["sns:*"], "Resource": "arn:aws:sns:*:*:*", "Effect": "Allow" },
{ "Action": ["sts:GetCallerIdentity"], "Resource": "*", "Effect": "Allow" }
]
}

Production (restricted to specific resources):

{
"Statement": [
{
"Action": ["sqs:CreateQueue", "sqs:GetQueueUrl", "sqs:GetQueueAttributes",
"sqs:SetQueueAttributes", "sqs:ReceiveMessage", "sqs:SendMessage",
"sqs:DeleteMessage", "sqs:ChangeMessageVisibility"],
"Resource": ["arn:aws:sqs:us-east-1:123456789012:orders.fifo"],
"Effect": "Allow"
},
{
"Action": ["sns:CreateTopic", "sns:GetTopicAttributes", "sns:Subscribe",
"sns:Publish"],
"Resource": ["arn:aws:sns:us-east-1:123456789012:order-events"],
"Effect": "Allow"
},
{
"Action": ["kms:Decrypt", "kms:Encrypt", "kms:GenerateDataKey", "kms:DescribeKey"],
"Resource": "arn:aws:kms:us-east-1:123456789012:key/your-key-id",
"Effect": "Allow"
},
{
"Action": ["sts:GetCallerIdentity"],
"Resource": "*",
"Effect": "Allow"
}
]
}

Cloud Best Practices

Use FIFO queues for ordered operations — commands that must be processed in sequence per entity
Use standard queues for independent operations — notifications, emails, analytics
Group related commands to the same queue — CreateOrder, UpdateOrder, CancelOrder all go to orders.fifo
Enable SQL-based idempotency in production — in-memory is insufficient for multi-instance deployments
Enable KMS encryption for sensitive data — PII, financial data, health records
Use infrastructure-as-code for production — CloudFormation/Terraform for queues and topics; let bootstrapper handle dev only
Monitor health checks and metrics — alert on sourceflow.aws.message.processing_error and circuit breaker state
Configure dead letter queues — review failed messages regularly

For detailed AWS configuration, IAM policies, and architecture diagrams, see the SourceFlow.Cloud.AWS Documentation.

Implementation Guide

Creating a Complete Feature

Let's implement a complete banking feature using SourceFlow.Net with Entity Framework persistence:

1. Define Domain Objects

// EntitypublicclassBankAccount:IEntity{publicintId{get;set;}publicstringAccountName{get;set;}publicdecimalBalance{get;set;}publicboolIsClosed{get;set;}publicDateTimeCreatedOn{get;set;}publicDateTimeActiveOn{get;set;}publicstringClosureReason{get;set;}}// Command PayloadspublicclassCreateAccountPayload:IPayload{publicintId{get;set;}publicstringAccountName{get;set;}publicdecimalInitialAmount{get;set;}}publicclassTransactionPayload:IPayload{publicintId{get;set;}publicdecimalAmount{get;set;}}

2. Create Commands

publicclassCreateAccount:Command<CreateAccountPayload>{// Parameterless constructor required for command deserialization from storepublicCreateAccount():base(){}publicCreateAccount(CreateAccountPayloadpayload):base(payload){}}publicclassDepositMoney:Command<TransactionPayload>{// Parameterless constructor required for command deserialization from storepublicDepositMoney():base(){}publicDepositMoney(TransactionPayloadpayload):base(payload){}}publicclassWithdrawMoney:Command<TransactionPayload>{// Parameterless constructor required for command deserialization from storepublicWithdrawMoney():base(){}publicWithdrawMoney(TransactionPayloadpayload):base(payload){}}

3. Define Events

publicclassAccountCreated:Event<BankAccount>{publicAccountCreated(BankAccountpayload):base(payload){}}publicclassMoneyDeposited:Event<BankAccount>{publicMoneyDeposited(BankAccountpayload):base(payload){}}publicclassMoneyWithdrawn:Event<BankAccount>{publicMoneyWithdrawn(BankAccountpayload):base(payload){}}

4. Implement Saga

publicclassAccountSaga:Saga<BankAccount>,IHandles<CreateAccount>,IHandles<DepositMoney>,IHandles<WithdrawMoney>{publicasyncTaskHandle(CreateAccountcommand){// Validationif(string.IsNullOrEmpty(command.Payload.AccountName))thrownewArgumentException("Account name is required");if(command.Payload.InitialAmount<=0)thrownewArgumentException("Initial amount must be positive");// Create entityvaraccount=newBankAccount{Id=command.Payload.Id,AccountName=command.Payload.AccountName,Balance=command.Payload.InitialAmount,CreatedOn=DateTime.UtcNow,ActiveOn=DateTime.UtcNow};// Persist to Entity Storeawaitrepository.Persist(account);// Raise eventawaitRaise(newAccountCreated(account));logger.LogInformation("Account created: {AccountId} for {Holder} with balance {Balance}",account.Id,account.AccountName,account.Balance);}publicasyncTaskHandle(DepositMoneycommand){varaccount=awaitrepository.Get<BankAccount>(command.Payload.Id);if(account.IsClosed)thrownewInvalidOperationException("Cannot deposit to closed account");account.Balance+=command.Payload.Amount;awaitrepository.Persist(account);awaitRaise(newMoneyDeposited(account));logger.LogInformation("Deposited {Amount} to account {AccountId}. New balance: {Balance}",command.Payload.Amount,account.Id,account.Balance);}publicasyncTaskHandle(WithdrawMoneycommand){varaccount=awaitrepository.Get<BankAccount>(command.Payload.Id);if(account.IsClosed)thrownewInvalidOperationException("Cannot withdraw from closed account");if(account.Balance<command.Payload.Amount)thrownewInvalidOperationException("Insufficient funds");account.Balance-=command.Payload.Amount;awaitrepository.Persist(account);awaitRaise(newMoneyWithdrawn(account));logger.LogInformation("Withdrew {Amount} from account {AccountId}. New balance: {Balance}",command.Payload.Amount,account.Id,account.Balance);}}

5. Create Aggregate

publicinterfaceIAccountAggregate:IAggregate{voidCreateAccount(intaccountId,stringholder,decimalamount);voidDeposit(intaccountId,decimalamount);voidWithdraw(intaccountId,decimalamount);}publicclassAccountAggregate:Aggregate<BankAccount>,IAccountAggregate{publicvoidCreateAccount(intaccountId,stringholder,decimalamount){Send(newCreateAccount(newCreateAccountPayload{Id=accountId,AccountName=holder,InitialAmount=amount}));}publicvoidDeposit(intaccountId,decimalamount){Send(newDepositMoney(newTransactionPayload{Id=accountId,Amount=amount}));}publicvoidWithdraw(intaccountId,decimalamount){Send(newWithdrawMoney(newTransactionPayload{Id=accountId,Amount=amount}));}}

6. Build Read Models

publicclassAccountViewModel:IViewModel{publicintId{get;set;}publicstringAccountName{get;set;}publicdecimalCurrentBalance{get;set;}publicDateTimeCreatedDate{get;set;}publicDateTimeLastUpdated{get;set;}publicintTransactionCount{get;set;}publicboolIsClosed{get;set;}publicstringClosureReason{get;set;}publicintVersion{get;set;}publicDateTimeActiveOn{get;set;}}publicclassAccountView:View,IProjectOn<AccountCreated>,IProjectOn<MoneyDeposited>,IProjectOn<MoneyWithdrawn>{publicasyncTaskApply(AccountCreated@event){varview=newAccountViewModel{Id=@event.Payload.Id,AccountName=@event.Payload.AccountName,CurrentBalance=@event.Payload.Balance,CreatedDate=@event.Payload.CreatedOn,LastUpdated=DateTime.UtcNow,TransactionCount=0,IsClosed=false,ActiveOn=@event.Payload.ActiveOn};awaitprovider.Push(view);logger.LogInformation("Created view model for account {AccountId}",view.Id);}publicasyncTaskApply(MoneyDeposited@event){varview=awaitprovider.Find<AccountViewModel>(@event.Payload.Id);view.CurrentBalance=@event.Payload.Balance;view.TransactionCount++;view.LastUpdated=DateTime.UtcNow;awaitprovider.Push(view);logger.LogInformation("Updated view model for account {AccountId} after deposit",view.Id);}publicasyncTaskApply(MoneyWithdrawn@event){varview=awaitprovider.Find<AccountViewModel>(@event.Payload.Id);view.CurrentBalance=@event.Payload.Balance;view.TransactionCount++;view.LastUpdated=DateTime.UtcNow;awaitprovider.Push(view);logger.LogInformation("Updated view model for account {AccountId} after withdrawal",view.Id);}}

7. Application Setup

// Program.csusingSourceFlow;usingSourceFlow.Stores.EntityFramework;usingSourceFlow.Stores.EntityFramework.Extensions;usingMicrosoft.Extensions.DependencyInjection;usingMicrosoft.Extensions.Logging;varservices=newServiceCollection();// Add loggingservices.AddLogging(builder =>{builder.AddConsole();builder.SetMinimumLevel(LogLevel.Information);});// Register entity and view model types BEFORE building service providerEntityDbContext.RegisterEntityType<BankAccount>();ViewModelDbContext.RegisterViewModelType<AccountViewModel>();// Configure SourceFlowservices.UseSourceFlow(typeof(Program).Assembly);// Add Entity Framework storesservices.AddSourceFlowEfStores("Server=localhost;Database=SourceFlow;Integrated Security=true;TrustServerCertificate=true;");varserviceProvider=services.BuildServiceProvider();// Ensure databases are created and migratedvarcommandContext=serviceProvider.GetRequiredService<CommandDbContext>();awaitcommandContext.Database.EnsureCreatedAsync();varentityContext=serviceProvider.GetRequiredService<EntityDbContext>();awaitentityContext.Database.EnsureCreatedAsync();entityContext.ApplyMigrations();varviewModelContext=serviceProvider.GetRequiredService<ViewModelDbContext>();awaitviewModelContext.Database.EnsureCreatedAsync();viewModelContext.ApplyMigrations();// Use the aggregatevaraggregateFactory=serviceProvider.GetRequiredService<IAggregateFactory>();varaccountAggregate=awaitaggregateFactory.Create<IAccountAggregate>();accountAggregate.CreateAccount(999,"John Doe",1000m);accountAggregate.Deposit(999,500m);accountAggregate.Withdraw(999,200m);// Give async processing time to completeawaitTask.Delay(500);// Query the read modelvarviewModelStore=serviceProvider.GetRequiredService<IViewModelStoreAdapter>();varaccountView=awaitviewModelStore.Find<AccountViewModel>(999);Console.WriteLine($"Account: {accountView.AccountName}");Console.WriteLine($"Balance: {accountView.CurrentBalance:C}");Console.WriteLine($"Transactions: {accountView.TransactionCount}");

For complete examples including ASP.NET Core, PostgreSQL, and production configurations, see EntityFramework Usage Examples.

Advanced Features

Event Replay

SourceFlow.Net provides built-in command replay functionality for debugging and state reconstruction:

varaccountAggregate=serviceProvider.GetRequiredService<IAccountAggregate>();// Replay all commands for an aggregateawaitaccountAggregate.ReplayHistory(accountId);// The framework automatically handles:// 1. Loading commands from store// 2. Marking commands as replay// 3. Re-executing command handlers// 4. Updating projections

Metadata and Auditing

Every command and event includes rich metadata to add producer and consumer centric custom properties.

publicinterfaceIMetadata{GuidEventId{get;set;}boolIsReplay{get;set;}DateTimeOccurredOn{get;set;}intSequenceNo{get;set;}IDictionary<string,object>Properties{get;set;}}

Store Adapters

SourceFlow provides high-level adapters for common operations:

// Entity Store AdapterpublicinterfaceIEntityStoreAdapter{TaskPersist<TEntity>(TEntityentity)whereTEntity:class,IEntity;Task<TEntity>Get<TEntity>(intid)whereTEntity:class,IEntity;}// ViewModel Store AdapterpublicinterfaceIViewModelStoreAdapter{TaskPush<TViewModel>(TViewModelmodel)whereTViewModel:class,IViewModel;Task<TViewModel>Find<TViewModel>(intid)whereTViewModel:class,IViewModel;}// Command Store AdapterpublicinterfaceICommandStoreAdapter{TaskCommit(ICommandcommand);Task<IEnumerable<ICommand>>Retrieve(intentityId);}

Performance and Observability

SourceFlow.Net includes comprehensive production-ready features for monitoring, fault tolerance, and high-performance scenarios.

OpenTelemetry Integration

Built-in support for distributed tracing and metrics collection at scale.

Features

Distributed Tracing: Automatically track command execution, event dispatching, and store operations
Metrics Collection: Monitor command rates, saga executions, entity creations, and operation durations
Multiple Exporters: Support for Console, OTLP (Jaeger, Zipkin), and custom exporters
Minimal Overhead: <1ms latency impact, <2% CPU overhead

Quick Setup

Development (Console Exporter):

usingSourceFlow.Observability;usingOpenTelemetry;varservices=newServiceCollection();// Enable observability with console outputservices.AddSourceFlowTelemetry(serviceName:"MyEventSourcedApp",serviceVersion:"1.0.0");services.AddOpenTelemetry().AddSourceFlowConsoleExporter();services.UseSourceFlow();

Production (OTLP Exporter):

services.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="ProductionApp";options.ServiceVersion="1.0.0";});services.AddOpenTelemetry().AddSourceFlowOtlpExporter("http://localhost:4317").AddSourceFlowResourceAttributes(("environment","production"),("region","us-east-1"));

Instrumented Operations

All core operations are automatically traced:

Command Operations:

sourceflow.commandbus.dispatch - Command dispatch and persistence
sourceflow.commanddispatcher.send - Command distribution to sagas
sourceflow.domain.command.append - Command persistence
sourceflow.domain.command.load - Command loading

Event Operations:

sourceflow.eventqueue.enqueue - Event queuing
sourceflow.eventdispatcher.dispatch - Event distribution

Store Operations:

sourceflow.entitystore.persist / get / delete - Entity operations
sourceflow.viewmodelstore.persist / find / delete - ViewModel operations

Metrics Collected

sourceflow.domain.commands.executed - Counter of executed commands
sourceflow.domain.sagas.executed - Counter of saga executions
sourceflow.domain.entities.created - Counter of entity creations
sourceflow.domain.operation.duration - Histogram of operation durations (ms)
sourceflow.domain.serialization.duration - Histogram of serialization performance

Integration with Existing Telemetry

services.AddOpenTelemetry().WithTracing(builder =>builder.AddSource("SourceFlow.Domain")// Add SourceFlow.AddAspNetCoreInstrumentation().AddHttpClientInstrumentation().AddOtlpExporter()).WithMetrics(builder =>builder.AddMeter("SourceFlow.Domain")// Add SourceFlow.AddAspNetCoreInstrumentation().AddPrometheusExporter());

ArrayPool Memory Optimization

Dramatically reduce memory allocations in high-throughput scenarios using ArrayPool<T>.

Performance Benefits

Memory Allocation Reduction:

Before: ~40MB allocations for 10,000 commands
After: <1MB allocations for 10,000 commands
Result: ~40x reduction in allocations

GC Pressure Reduction:

Gen 0 Collections: ↓70%
Gen 1 Collections: ↓50%
Gen 2 Collections: ↓30%

Throughput Improvements:

Command Throughput: +25-40%
Event Dispatching: +30-50%
Serialization: +20-35%

Features

Task Buffer Pooling: Reduces allocations in parallel task execution
JSON Serialization Pooling: Reuses byte buffers for JSON operations
Zero Configuration: Works automatically, no code changes required
Production Tested: Optimized for extreme throughput scenarios

Optimized Components

TaskBufferPool:

Pools task arrays for parallel execution
Used in CommandDispatcher and EventDispatcher
Automatic buffer rental and return

ByteArrayPool:

Pools byte arrays for JSON serialization
Used in CommandStoreAdapter
Custom IBufferWriter<byte> implementation

ArrayPool optimizations are automatically applied to:

Command serialization/deserialization
Event dispatching (parallel task execution)
Store adapter operations

Resilience with Polly (Entity Framework)

The Entity Framework integration includes Polly-based resilience patterns for fault tolerance.

Features

Retry Policy: Automatic retry with exponential backoff and jitter
Circuit Breaker: Prevents cascading failures
Timeout Policy: Enforces maximum execution time

Configuration

services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=connectionString;// Enable resilienceoptions.Resilience.Enabled=true;// Retry configurationoptions.Resilience.Retry.MaxRetryAttempts=3;options.Resilience.Retry.BaseDelayMs=1000;options.Resilience.Retry.UseExponentialBackoff=true;options.Resilience.Retry.UseJitter=true;// Circuit breaker configurationoptions.Resilience.CircuitBreaker.Enabled=true;options.Resilience.CircuitBreaker.FailureThreshold=5;options.Resilience.CircuitBreaker.BreakDurationMs=30000;// Timeout configurationoptions.Resilience.Timeout.Enabled=true;options.Resilience.Timeout.TimeoutMs=30000;});

Benefits

Transient Failure Handling: Automatically recovers from temporary issues
Cascading Failure Prevention: Circuit breaker stops calling failing services
Resource Protection: Timeouts prevent hanging operations
Self-Healing: System automatically recovers when service becomes available

Entity Framework Observability

Additional observability features specific to Entity Framework stores.

Configuration

services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=connectionString;// Configure observabilityoptions.Observability.Enabled=true;options.Observability.ServiceName="MyApplication";options.Observability.ServiceVersion="1.0.0";// Tracing configurationoptions.Observability.Tracing.Enabled=true;options.Observability.Tracing.TraceDatabaseOperations=true;options.Observability.Tracing.IncludeSqlInTraces=false;// Enable for debuggingoptions.Observability.Tracing.SamplingRatio=0.1;// Sample 10% in production// Metrics configurationoptions.Observability.Metrics.Enabled=true;options.Observability.Metrics.CollectDatabaseMetrics=true;});// Configure exportersservices.AddOpenTelemetry().WithTracing(tracing =>tracing.AddSource("SourceFlow.EntityFramework").AddEntityFrameworkCoreInstrumentation().AddJaegerExporter()).WithMetrics(metrics =>metrics.AddMeter("SourceFlow.EntityFramework").AddPrometheusExporter());

Additional Traces

sourceflow.ef.command.append - EF command storage
sourceflow.ef.command.load - EF command loading
sourceflow.ef.entity.persist - EF entity persistence
sourceflow.ef.viewmodel.persist - EF view model persistence

Additional Metrics

sourceflow.commands.appended - EF command append counter
sourceflow.commands.loaded - EF command load counter
sourceflow.entities.persisted - EF entity persistence counter
sourceflow.viewmodels.persisted - EF view model persistence counter
sourceflow.database.connections - Active connection gauge

Production Configuration Examples

Development:

services.AddSourceFlowTelemetry("DevApp","1.0.0");services.AddOpenTelemetry().AddSourceFlowConsoleExporter();services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString="Data Source=dev.db";options.Resilience.Enabled=false;// Easier debuggingoptions.Observability.Enabled=true;options.Observability.Tracing.IncludeSqlInTraces=true;options.Observability.Tracing.SamplingRatio=1.0;// Trace everything});

Production:

services.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="ProductionApp";options.ServiceVersion="1.0.0";});services.AddOpenTelemetry().AddSourceFlowOtlpExporter(otlpEndpoint).AddSourceFlowResourceAttributes(("environment","production"),("deployment.region",region));services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=connectionString;// Production resilience settingsoptions.Resilience.Enabled=true;options.Resilience.Retry.MaxRetryAttempts=3;options.Resilience.CircuitBreaker.Enabled=true;options.Resilience.CircuitBreaker.FailureThreshold=10;// Production observability settingsoptions.Observability.Enabled=true;options.Observability.Tracing.IncludeSqlInTraces=false;options.Observability.Tracing.SamplingRatio=0.1;// Sample 10%});

High-Throughput:

services.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="HighThroughputApp";});services.AddOpenTelemetry().AddSourceFlowOtlpExporter(otlpEndpoint).ConfigureSourceFlowBatchProcessing(maxQueueSize:2048,maxExportBatchSize:512,scheduledDelayMilliseconds:5000);services.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=connectionString;// Optimized for throughputoptions.Resilience.Enabled=true;options.Resilience.Retry.MaxRetryAttempts=2;options.Resilience.Retry.BaseDelayMs=500;// Reduced overheadoptions.Observability.Enabled=true;options.Observability.Tracing.SamplingRatio=0.01;// Sample 1%});// ArrayPool optimizations are automatically applied

Monitoring Dashboard Queries

Use these queries in Grafana/Prometheus:

Average Command Processing Time:

rate(sourceflow_domain_operation_duration_sum{operation="sourceflow.commandbus.dispatch"}[5m])
/ rate(sourceflow_domain_operation_duration_count{operation="sourceflow.commandbus.dispatch"}[5m])

Command Throughput:

rate(sourceflow_domain_commands_executed[5m])

Serialization Performance (P95):

histogram_quantile(0.95,
rate(sourceflow_domain_serialization_duration_bucket[5m])
)

Package Dependencies

Core SourceFlow:

OpenTelemetry (1.14.0)
OpenTelemetry.Api (1.14.0)
OpenTelemetry.Exporter.Console (1.14.0)
OpenTelemetry.Exporter.OpenTelemetryProtocol (1.14.0)
OpenTelemetry.Extensions.Hosting (1.14.0)
Microsoft.Extensions.DependencyInjection.Abstractions (10.0.0)
Microsoft.Extensions.Logging.Abstractions (10.0.0)

Entity Framework Stores:

Microsoft.EntityFrameworkCore (9.0.0)
Polly (8.4.2) - For resilience patterns
OpenTelemetry.Instrumentation.EntityFrameworkCore (1.0.0-beta.12)

All packages are free from known vulnerabilities (as of November 2025).

Additional Resources

OpenTelemetry Documentation: https://opentelemetry.io/docs/
ArrayPool Documentation: https://docs.microsoft.com/en-us/dotnet/api/system.buffers.arraypool-1
Polly Documentation: https://github.com/App-vNext/Polly
SourceFlow.Net OBSERVABILITY_AND_PERFORMANCE.md: Detailed performance documentation
SourceFlow.Stores.EntityFramework ENHANCEMENTS.md: EF-specific enhancements guide

Best Practices

1. Command Design

Always include a parameterless constructor for serialization support:

// ✅ Good: Specific, intention-revealing commands with proper constructorspublicclassWithdrawMoney:Command<WithdrawPayload>{// Required for deserialization from command storepublicWithdrawMoney():base(){}publicWithdrawMoney(WithdrawPayloadpayload):base(payload){}}publicclassDepositMoney:Command<DepositPayload>{// Required for deserialization from command storepublicDepositMoney():base(){}publicDepositMoney(DepositPayloadpayload):base(payload){}}// ❌ Bad: Generic, unclear commandspublicclassUpdateAccount:Command<AccountPayload>{}// ❌ Bad: Missing parameterless constructorpublicclassTransferMoney:Command<TransferPayload>{publicTransferMoney(TransferPayloadpayload):base(payload){}// This command cannot be deserialized from the command store!}

Key Requirements:

Use specific, intention-revealing names
Always include a public parameterless constructor
Include a constructor that accepts the payload
Keep commands immutable after creation

2. Event Granularity

// ✅ Good: Fine-grained, specific eventspublicclassAccountCreated:Event<BankAccount>{}publicclassAccountCredited:Event<BankAccount>{}publicclassAccountDebited:Event<BankAccount>{}// ❌ Bad: Coarse-grained, generic eventspublicclassAccountChanged:Event<BankAccount>{}

3. Saga Responsibility

// ✅ Good: Single responsibilitypublicclassAccountSaga:Saga<BankAccount>,IHandles<CreateAccount>,IHandles<CloseAccount>{// Handles account lifecycle only}// ❌ Bad: Multiple responsibilitiespublicclassMegaSaga:Saga<BankAccount>,IHandles<CreateAccount>,IHandles<ProcessLoan>,IHandles<SendEmail>{// Too many responsibilities}

4. Type Registration

// ✅ Good: Register types early, before building service providerEntityDbContext.RegisterAssembly(typeof(BankAccount).Assembly);ViewModelDbContext.RegisterAssembly(typeof(AccountViewModel).Assembly);varserviceProvider=services.BuildServiceProvider();// Apply migrations after database creationentityContext.Database.EnsureCreated();entityContext.ApplyMigrations();// ❌ Bad: Relying solely on auto-discoveryvarserviceProvider=services.BuildServiceProvider();// Types may not be discovered reliably

5. Error Handling

publicclassAccountSaga:Saga<BankAccount>{publicasyncTaskHandle(WithdrawMoneycommand){try{varaccount=awaitrepository.Get<BankAccount>(command.Payload.Id);// Validate business rulesif(account.IsClosed)thrownewAccountClosedException($"Account {account.Id} is closed");if(account.Balance<command.Payload.Amount)thrownewInsufficientFundsException($"Insufficient funds in account {account.Id}");// Process transactionaccount.Balance-=command.Payload.Amount;awaitrepository.Persist(account);awaitRaise(newMoneyWithdrawn(account));}catch(Exceptionex){logger.LogError(ex,"Failed to process withdrawal for account {AccountId}",command.Payload.Id);// Publish failure event if neededawaitRaise(newWithdrawalFailed(newWithdrawalFailureDetails{AccountId=command.Payload.Id,Amount=command.Payload.Amount,Reason=ex.Message}));throw;}}}

6. Database Migrations

// ✅ Good: Use ApplyMigrations for dynamic typesentityContext.Database.EnsureCreated();entityContext.ApplyMigrations();// Creates tables for registered typesviewModelContext.Database.EnsureCreated();viewModelContext.ApplyMigrations();// Creates tables for view models// For production: Use EF Core migrations for static schema// dotnet ef migrations add InitialCreate// dotnet ef database update

7. Production Monitoring

// ✅ Good: Enable observability in productionservices.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="ProductionApp";options.ServiceVersion=Assembly.GetExecutingAssembly().GetName().Version.ToString();});// Configure appropriate sampling rateservices.AddOpenTelemetry().AddSourceFlowOtlpExporter(otlpEndpoint).AddSourceFlowResourceAttributes(("environment",Environment.GetEnvironmentVariable("ENVIRONMENT")));// ❌ Bad: No observability in productionservices.UseSourceFlow();// Can't diagnose performance issues or failures

8. Resilience Configuration

// ✅ Good: Enable resilience for production databasesservices.AddSourceFlowEfStores(options =>{options.DefaultConnectionString=connectionString;options.Resilience.Enabled=true;options.Resilience.Retry.MaxRetryAttempts=3;options.Resilience.CircuitBreaker.Enabled=true;});// ❌ Bad: No resilience (will fail on transient errors)services.AddSourceFlowEfStores(connectionString);

9. Command Serialization Requirements

// ✅ Good: Command with parameterless constructorpublicclassProcessPayment:Command<PaymentPayload>{// REQUIRED: Parameterless constructor for deserializationpublicProcessPayment():base(){}publicProcessPayment(PaymentPayloadpayload):base(payload){}}// ❌ Bad: Missing parameterless constructorpublicclassProcessPayment:Command<PaymentPayload>{publicProcessPayment(PaymentPayloadpayload):base(payload){}// Will throw MissingMethodException during command replay!}// ✅ Good: Payload classes don't need parameterless constructorspublicclassPaymentPayload:IPayload{publicintId{get;set;}publicdecimalAmount{get;set;}publicstringCurrency{get;set;}}

Important Notes:

Commands MUST have a public parameterless constructor for deserialization from the command store
Payload classes use property setters for deserialization (no parameterless constructor required)
Without a parameterless constructor, command replay and aggregate reconstruction will fail
The Entity Framework CommandStoreAdapter uses reflection to recreate command instances

FAQ

Q: How does SourceFlow.Net handle persistence?

A: SourceFlow.Net uses a store abstraction pattern with multiple implementation options:

In-Memory Stores: Built-in for testing and prototyping
Entity Framework Stores: Production-ready with support for SQL Server, PostgreSQL, SQLite, etc.
Custom Stores: Implement ICommandStore, IEntityStore, and IViewModelStore for your own persistence

Q: Can I use different databases for commands, entities, and view models?

A: Yes! The Entity Framework integration supports separate databases:

services.AddSourceFlowEfStoresWithCustomProviders(commandContextConfig: options =>options.UseSqlServer("..."),entityContextConfig: options =>options.UsePostgreSql("..."),viewModelContextConfig: options =>options.UseSqlite("..."));

Q: How do I handle dynamic entity and view model types?

A: Use the type registration and migration system:

Register types before building the service provider
Call EnsureCreated() to create the base schema
Call ApplyMigrations() to create tables for registered types

EntityDbContext.RegisterEntityType<MyEntity>();ViewModelDbContext.RegisterViewModelType<MyViewModel>();// Build service provider...entityContext.Database.EnsureCreated();entityContext.ApplyMigrations();

Q: Why do my commands need a parameterless constructor?

A: The CommandStoreAdapter uses reflection to deserialize commands from the database. When replaying commands, it needs to create instances without knowing the payload in advance.

Required pattern:

publicclassCreateAccount:Command<CreateAccountPayload>{// Required for deserializationpublicCreateAccount():base(){}// Used for creating new commandspublicCreateAccount(CreateAccountPayloadpayload):base(payload){}}

Without the parameterless constructor, command replay will fail with a MissingMethodException.

Q: What database providers are supported?

A: SourceFlow.Stores.EntityFramework supports any EF Core provider:

SQL Server (default)
PostgreSQL (via Npgsql.EntityFrameworkCore.PostgreSQL)
SQLite (via Microsoft.EntityFrameworkCore.Sqlite)
MySQL (via Pomelo.EntityFrameworkCore.MySql)
In-Memory (via Microsoft.EntityFrameworkCore.InMemoryDatabase)
And more...

Q: How do I test with SourceFlow.Net?

A: Use in-memory databases for fast, isolated tests:

[SetUp]publicvoidSetup(){EntityDbContext.RegisterEntityType<TestEntity>();ViewModelDbContext.RegisterViewModelType<TestViewModel>();varconnection=newSqliteConnection("DataSource=:memory:");connection.Open();services.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseSqlite(connection));// Build and setup...}

Q: Why use the "T" prefix for table names?

A: The "T" prefix distinguishes dynamically created tables from EF Core's built-in tables, making it clear which tables are part of your domain model versus infrastructure.

Q: Should I enable observability in production?

A: Yes! Observability has minimal overhead (<1ms latency, <2% CPU) and provides invaluable insights:

Distributed tracing helps debug issues across services
Metrics help identify performance bottlenecks
Sampling (10%) provides good coverage with minimal cost

services.AddSourceFlowTelemetry(options =>{options.Enabled=true;options.ServiceName="ProductionApp";});services.AddOpenTelemetry().AddSourceFlowOtlpExporter(otlpEndpoint);

Q: When should I use resilience patterns?

A: Always enable resilience in production for database operations:

Retry policies handle transient network failures
Circuit breakers prevent cascading failures
Timeouts prevent hanging operations

services.AddSourceFlowEfStores(options =>{options.Resilience.Enabled=true;});

Q: How much does ArrayPool improve performance?

A: In high-throughput scenarios (>1000 commands/second):

Memory: 40x reduction in allocations (40MB → <1MB for 10K commands)
GC: 70% reduction in Gen0 collections
Throughput: 25-40% improvement
Zero configuration: Works automatically once enabled

ArrayPool optimizations are built-in and automatically applied to command serialization and event dispatching.

Q: How do I handle schema changes?

A: For production applications:

Use EF Core migrations for base schema
Use ApplyMigrations() for dynamic types
Version your entities and view models
Implement upcasting for old events

For development/testing:

Use Database.EnsureDeleted() and EnsureCreated()
Use in-memory databases that reset on each test

Q: Can I use SourceFlow.EntityFramework with MySQL or other databases?

A: Yes! Use AddSourceFlowEfStoresWithCustomProvider for any EF Core supported database:

// MySQLvarserverVersion=newMySqlServerVersion(newVersion(8,0,21));services.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseMySql(connectionString,serverVersion));// SQLiteservices.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseSqlite("Data Source=sourceflow.db"));// PostgreSQLservices.AddSourceFlowEfStoresWithCustomProvider(options =>options.UseNpgsql(connectionString));

The AddSourceFlowEfStores methods without "CustomProvider" use SQL Server by default.

Q: How do I add AWS cloud messaging to my application?

A: Install SourceFlow.Cloud.AWS and configure using the fluent API:

dotnetadd package SourceFlow.Cloud.AWS

services.UseSourceFlowAws(
options =>{options.Region=RegionEndpoint.USEast1;},
bus =>bus.Send.Command<MyCommand>(q =>q.Queue("my-queue.fifo")).Raise.Event<MyEvent>(t =>t.Topic("my-events")).Listen.To.CommandQueue("my-queue.fifo").Subscribe.To.Topic("my-events"));

The bootstrapper automatically provisions SQS queues, SNS topics, and subscriptions at startup.

Q: Do I need to create SQS queues and SNS topics manually?

A: No. The AwsBusBootstrapper runs as an IHostedService and creates all required resources at startup. All operations are idempotent. For production, consider using CloudFormation/Terraform for resource management and letting the bootstrapper verify they exist.

Q: How do I prevent duplicate message processing in multi-instance deployments?

A: Use SQL-based idempotency from SourceFlow.Stores.EntityFramework:

services.AddSourceFlowIdempotency(connectionString:configuration.GetConnectionString("IdempotencyStore"),cleanupIntervalMinutes:60);

This uses database transactions for thread-safe duplicate detection across instances. For single-instance deployments, the default InMemoryIdempotencyService is sufficient.

Q: How do I test AWS integrations locally?

A: Use LocalStack with the provided scripts:

./tests/SourceFlow.Cloud.AWS.Tests/run-integration-tests.ps1 # Windows
./tests/SourceFlow.Cloud.AWS.Tests/run-integration-tests.sh # Linux/macOS

Or manually start LocalStack via Docker and set AWS_ENDPOINT_URL=http://localhost:4566.

Q: What's the difference between EnsureCreated() and ApplyMigrations()?

A:

EnsureCreated(): Creates the database and base schema (Commands, fixed tables)
ApplyMigrations(): Creates tables for dynamically registered entities and view models

Always call both in the correct order:

awaitentityContext.Database.EnsureCreatedAsync();// Create databaseentityContext.ApplyMigrations();// Create dynamic tables

Q: How do I configure EF Core 9.0 for testing to avoid provider conflicts?

A: When testing with multiple DbContext providers (e.g., SQLite for tests, SQL Server for production), use EnableServiceProviderCaching(false):

services.AddDbContext<EntityDbContext>(options =>options.UseSqlite(connection).EnableServiceProviderCaching(false));// Required for EF Core 9.0

This prevents the "multiple provider" error when using different providers in the same service collection.

Q: Should I register stores manually or use AddSourceFlowEfStores?

A: Use AddSourceFlowEfStores for production. Only register manually for special cases like:

Testing scenarios requiring specific service configuration
Custom implementations of resilience or telemetry services
Avoiding provider conflicts in test setups

Example manual registration:

varefOptions=newSourceFlowEfOptions();services.AddSingleton(efOptions);services.AddScoped<IDatabaseResiliencePolicy,DatabaseResiliencePolicy>();services.AddScoped<IDatabaseTelemetryService,DatabaseTelemetryService>();services.AddScoped<ICommandStore,EfCommandStore>();services.AddScoped<IEntityStore,EfEntityStore>();services.AddScoped<IViewModelStore,EfViewModelStore>();

Q: How do table naming conventions affect my database schema?

A: Table naming conventions transform entity type names into table names:

// Default (PascalCase, no prefix/suffix)BankAccount → BankAccount// Snake case with pluralizationoptions.EntityTableNaming.Casing=TableNameCasing.SnakeCase;options.EntityTableNaming.Pluralize=true;BankAccount → bank_accounts// With schemaoptions.EntityTableNaming.UseSchema=true;options.EntityTableNaming.SchemaName="domain";BankAccount → domain.BankAccount// Combined
BankAccount → domain.bank_accounts

Set naming conventions BEFORE calling ApplyMigrations() to ensure tables are created with the correct names.

Production Considerations

Performance Optimization

Use Separate Databases: Split command, entity, and view model stores across different databases
Enable Connection Pooling: Configure appropriate connection pool sizes
Optimize Queries: Use AsNoTracking() for read-only queries
Batch Operations: Use bulk insert/update operations where applicable
Enable ArrayPool: Automatically enabled for high-throughput scenarios (40x reduction in allocations)
Configure Observability: Use appropriate sampling rates for production (1-10%)
Enable Resilience: Use Polly policies for fault tolerance in production

Cloud Deployment

Enable SQL-based idempotency: Required for multi-instance deployments processing shared queues
Enable KMS encryption: For messages containing sensitive data (PII, financial, health)
Use FIFO queues: For commands requiring ordered processing per entity
Configure dead letter queues: Monitor and reprocess failed messages
Restrict IAM permissions: Scope SQS/SNS/KMS access to specific resource ARNs
Monitor cloud health checks: AwsHealthCheck validates SQS, SNS, and KMS connectivity
Use infrastructure-as-code: CloudFormation or Terraform for production AWS resources

Monitoring

// Health checks (database + AWS)services.AddHealthChecks().AddDbContextCheck<CommandDbContext>("commandstore").AddDbContextCheck<EntityDbContext>("entitystore").AddDbContextCheck<ViewModelDbContext>("viewmodelstore").AddCheck<AwsHealthCheck>("aws");// SQS, SNS, KMS connectivity// OpenTelemetry metrics and tracingservices.AddSourceFlowTelemetry("ProductionApp","1.0.0");services.AddOpenTelemetry().AddSourceFlowOtlpExporter("http://localhost:4317");// Monitor key metrics:// - Command throughput: sourceflow.domain.commands.executed// - Operation latency: sourceflow.domain.operation.duration (P50/P95/P99)// - Circuit breaker state: polly.circuit_breaker.state// - GC pressure: dotnet.gc.collections (reduced with ArrayPool)// - AWS dispatch: sourceflow.aws.command.dispatched / dispatch_duration// - AWS errors: sourceflow.aws.message.processing_error// - AWS events: sourceflow.aws.event.published / publish_duration

Deployment

Migrations: Apply EF Core migrations during deployment
Connection Strings: Use environment-specific configuration
Logging: Configure appropriate logging levels
Error Handling: Implement global exception handling

Common Issues and Solutions

Command Deserialization Failures:

Symptom: MissingMethodException or InvalidOperationException during command replay
Cause: Command class missing parameterless constructor
Solution: Add public parameterless constructor to all command classes

// Fix: Add parameterless constructorpublicclassMyCommand:Command<MyPayload>{publicMyCommand():base(){}// Required!publicMyCommand(MyPayloadpayload):base(payload){}}

Entity Tracking Conflicts:

Symptom: "Instance already being tracked" errors
Cause: Multiple entity instances with same ID in EF change tracker
Solution: Use AsNoTracking() for read operations or detach entities after save

EF Core 9.0 Provider Conflicts (Testing):

Symptom: "An error occurred accessing the database provider factory"
Cause: Multiple providers registered in same service collection
Solution: Use EnableServiceProviderCaching(false) in test configurations

Migration Failures:

Symptom: Tables not created for entities or view models
Cause: Types not registered before calling ApplyMigrations()
Solution: Register types using EntityDbContext.RegisterAssembly() before building service provider

Community and Support

Resources

GitHub Repository: https://github.com/CodeShayk/SourceFlow.Net
Documentation: https://github.com/CodeShayk/SourceFlow.Net/wiki
AWS Cloud Documentation: SourceFlow.Cloud.AWS README
Entity Framework Documentation: SourceFlow.Stores.EntityFramework README
Issues: https://github.com/CodeShayk/SourceFlow.Net/issues
Discussions: https://github.com/CodeShayk/SourceFlow.Net/discussions

License

SourceFlow.Net is released under the MIT License, making it free for both commercial and open-source use.

Conclusion

SourceFlow.Net provides a robust, scalable foundation for building event-sourced applications with .NET. By combining Event Sourcing, Domain-Driven Design, and CQRS patterns with flexible Entity Framework persistence and cloud-native AWS messaging, it enables developers to create maintainable, auditable, and performant distributed systems.

Start your journey to build better software with events as your foundation!

🤝Thank you!

🎯Start your journey with SourceFlow.Net today!

Uh oh!

Home

SourceFlow.Net - Complete Guide

Table of Contents

Introduction

What Makes SourceFlow.Net Special?

Key Features

Core Concepts

Event Sourcing

Key Benefits:

Example in SourceFlow.Net:

Domain-Driven Design (DDD)

Core DDD Elements in SourceFlow.Net:

Command Query Responsibility Segregation (CQRS)

Commands: Represent intent to change state

Queries: Handled through optimized view models

Projections: Update read models based on events

Architecture Overview

High-Level Architecture

Component Interactions

Cloud-Native Extension

Getting Started

Installation

Basic Setup

Framework Components

1. Aggregates

2. Sagas

3. Command Bus

4. Event Queue

5. Stores (Persistence Layer)

ICommandStore

IEntityStore

IViewModelStore

Persistence with Entity Framework

Features

Installation

Configuration Options

1. Single Connection String (All Stores)

2. Separate Connection Strings

3. Configuration-Based

4. Options Pattern

5. Custom Database Provider

6. Different Providers Per Store

Dynamic Type Registration

1. Explicit Type Registration

2. Assembly Scanning

3. Auto-Discovery (Fallback)

Table Naming Convention

Migration Helper

Complete Setup Example

Testing with In-Memory Database

EntityFramework Usage Examples

Example 1: Simple Console Application with SQL Server

Example 2: ASP.NET Core Web API with PostgreSQL

Example 3: Microservices with Separate Databases

Example 4: Production Configuration with Resilience and Observability

Example 5: Custom Table Naming with Schema Organization

Example 6: Background Service Processing Commands

Example 7: Multi-Tenant Setup with Database Per Tenant

Cloud Integration (AWS)

Installation

Quick Start

Bus Configuration API

Send Commands to SQS Queues

Raise Events to SNS Topics

Listen to Command Queues

Subscribe to Event Topics

Configuration Options

Message Encryption (KMS)

Automatic Resource Provisioning

Idempotency

In-Memory (Single Instance)

SQL-Based (Multi-Instance)

Command and Event Flow

Health Checks

Observability

Local Development with LocalStack

Quick Start (Recommended)

Manual Setup

Environment Variables