Example of implementing CQRS for a fraud detection system in a .NET Core application.
This example will demonstrate how to separate the command and query responsibilities and integrate event sourcing for real-time fraud detection.
Step 1: Define the Models
Define the models for commands and queries.
public class Transaction
{
public Guid Id { get; set; }
public decimal Amount { get; set; }
public DateTime Timestamp { get; set; }
public string UserId { get; set; }
public string Location { get; set; }
}
public class FraudAlert
{
public Guid Id { get; set; }
public string UserId { get; set; }
public string Message { get; set; }
public DateTime DetectedAt { get; set; }
}
Step 2: Command Side - Handling Transactions
Create a command handler to process transactions.
public class TransactionCommandHandler
{
private readonly IEventStore _eventStore;
public TransactionCommandHandler(IEventStore eventStore)
{
_eventStore = eventStore;
}
public async Task HandleAsync(Transaction transaction)
{
await _eventStore.SaveEventAsync(new TransactionEvent
{
Id = transaction.Id,
Amount = transaction.Amount,
Timestamp = transaction.Timestamp,
UserId = transaction.UserId,
Location = transaction.Location
});
}
}
Step 3: Event Store Interface
Define an interface for the event store.
public interface IEventStore
{
Task SaveEventAsync<T>(T @event) where T : class;
Task<IEnumerable<T>> GetEventsAsync<T>() where T : class;
}
Step 4: Query Side - Detecting Fraud
Create a query handler to detect fraud based on transaction events.
public class FraudDetectionQueryHandler
{
private readonly IEventStore _eventStore;
public FraudDetectionQueryHandler(IEventStore eventStore)
{
_eventStore = eventStore;
}
public async Task<IEnumerable<FraudAlert>> DetectFraudAsync(string userId)
{
var events = await _eventStore.GetEventsAsync<TransactionEvent>();
var userEvents = events.Where(e => e.UserId == userId).OrderBy(e => e.Timestamp).ToList();
var fraudAlerts = new List<FraudAlert>();
for (int i = 0; i < userEvents.Count - 1; i++)
{
var currentEvent = userEvents[i];
var nextEvent = userEvents[i + 1];
if (currentEvent.Location != nextEvent.Location && (nextEvent.Timestamp - currentEvent.Timestamp).TotalMinutes < 30)
{
fraudAlerts.Add(new FraudAlert
{
Id = Guid.NewGuid(),
UserId = userId,
Message = "Suspicious activity detected: multiple transactions from different locations within a short time frame.",
DetectedAt = DateTime.UtcNow
});
}
}
return fraudAlerts;
}
}
Step 5: Implementing the Event Store
Implement a simple in-memory event store for demonstration purposes.
public class InMemoryEventStore : IEventStore
{
private readonly List<object> _events = new List<object>();
public Task SaveEventAsync<T>(T @event) where T : class
{
_events.Add(@event);
return Task.CompletedTask;
}
public Task<IEnumerable<T>> GetEventsAsync<T>() where T : class
{
var events = _events.OfType<T>();
return Task.FromResult(events);
}
}
Step 6: Wiring Up the Application
Configure the services and middleware in Startup.cs.
public void ConfigureServices(IServiceCollection services)
{
services.AddSingleton<IEventStore, InMemoryEventStore>();
services.AddTransient<TransactionCommandHandler>();
services.AddTransient<FraudDetectionQueryHandler>();
services.AddControllers();
}
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
if (env.IsDevelopment())
{
app.UseDeveloperExceptionPage();
}
app.UseHttpsRedirection();
app.UseRouting();
app.UseEndpoints(endpoints =>
{
endpoints.MapControllers();
});
}
Step 7: Using the Handlers
Example usage of the command and query handlers.
public class TransactionsController : ControllerBase
{
private readonly TransactionCommandHandler _commandHandler;
private readonly FraudDetectionQueryHandler _queryHandler;
public TransactionsController(TransactionCommandHandler commandHandler, FraudDetectionQueryHandler queryHandler)
{
_commandHandler = commandHandler;
_queryHandler = queryHandler;
}
[HttpPost("transactions")]
public async Task<IActionResult> CreateTransaction([FromBody] Transaction transaction)
{
await _commandHandler.HandleAsync(transaction);
return Ok();
}
[HttpGet("fraud-alerts/{userId}")]
public async Task<IActionResult> GetFraudAlerts(string userId)
{
var alerts = await _queryHandler.DetectFraudAsync(userId);
return Ok(alerts);
}
}
This example demonstrates a basic implementation of CQRS for fraud detection. The command side handles transaction recording, while the query side analyzes these transactions to detect potential fraud. This separation allows for optimized processing and querying, making the system more efficient and scalable.
Example with a real database, you can replace the in-memory event store with a database-backed implementation.
Here, I’ll show you how to use Entity Framework Core with SQL Server for this purpose.
Step 1: Install Required Packages
First, install the necessary NuGet packages:
dotnet add package Microsoft.EntityFrameworkCore
dotnet add package Microsoft.EntityFrameworkCore.SqlServer
dotnet add package Microsoft.EntityFrameworkCore.Tools
Step 2: Define the Database Context
Create a DbContext for managing the database operations.
public class ApplicationDbContext : DbContext
{
public ApplicationDbContext(DbContextOptions<ApplicationDbContext> options) : base(options) { }
public DbSet<TransactionEvent> TransactionEvents { get; set; }
public DbSet<FraudAlert> FraudAlerts { get; set; }
}
Step 3: Update the Event Store Implementation
Implement the event store using Entity Framework Core.
public class EfEventStore : IEventStore
{
private readonly ApplicationDbContext _context;
public EfEventStore(ApplicationDbContext context)
{
_context = context;
}
public async Task SaveEventAsync<T>(T @event) where T : class
{
await _context.Set<T>().AddAsync(@event);
await _context.SaveChangesAsync();
}
public async Task<IEnumerable<T>> GetEventsAsync<T>() where T : class
{
return await _context.Set<T>().ToListAsync();
}
}
Step 4: Configure the Database in Startup.cs
Update the Startup.cs to configure the database context and use the new event store.
public void ConfigureServices(IServiceCollection services)
{
services.AddDbContext<ApplicationDbContext>(options =>
options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection")));
services.AddScoped<IEventStore, EfEventStore>();
services.AddTransient<TransactionCommandHandler>();
services.AddTransient<FraudDetectionQueryHandler>();
services.AddControllers();
}
Step 5: Update the Configuration
Add the connection string to your appsettings.json.
{
"ConnectionStrings": {
"DefaultConnection": "Server=(localdb)\\mssqllocaldb;Database=FraudDetectionDb;Trusted_Connection=True;MultipleActiveResultSets=true"
}
}
Step 6: Create the Database Migrations
Run the following commands to create and apply the database migrations.
dotnet ef migrations add InitialCreate
dotnet ef database update
Step 7: Update the Models for EF Core
Ensure your models are compatible with EF Core.
public class TransactionEvent
{
public Guid Id { get; set; }
public decimal Amount { get; set; }
public DateTime Timestamp { get; set; }
public string UserId { get; set; }
public string Location { get; set; }
}
public class FraudAlert
{
public Guid Id { get; set; }
public string UserId { get; set; }
public string Message { get; set; }
public DateTime DetectedAt { get; set; }
}
Step 8: Using the Handlers
The usage of the command and query handlers remains the same as before.
public class TransactionsController : ControllerBase
{
private readonly TransactionCommandHandler _commandHandler;
private readonly FraudDetectionQueryHandler _queryHandler;
public TransactionsController(TransactionCommandHandler commandHandler, FraudDetectionQueryHandler queryHandler)
{
_commandHandler = commandHandler;
_queryHandler = queryHandler;
}
[HttpPost("transactions")]
public async Task<IActionResult> CreateTransaction([FromBody] Transaction transaction)
{
await _commandHandler.HandleAsync(transaction);
return Ok();
}
[HttpGet("fraud-alerts/{userId}")]
public async Task<IActionResult> GetFraudAlerts(string userId)
{
var alerts = await _queryHandler.DetectFraudAsync(userId);
return Ok(alerts);
}
}
By following these steps, you can integrate the CQRS pattern with a real database using Entity Framework Core and SQL Server. This setup will allow you to handle real-time transactions and perform fraud detection efficiently.
Synchronizing data between two different databases for command and query operations in a CQRS
(Command Query Responsibility Segregation) setup can be achieved using several strategies.
Here are some common approaches:
1. Event Sourcing
Event sourcing is a powerful pattern where all changes to the application state are stored as a sequence of events. These events can then be used to update both the command and query databases.
Example
- Command Side: When a transaction occurs, an event is created and stored.
- Event Store: The event is saved in an event store.
- Event Handlers: Event handlers listen for these events and update the query database accordingly.
public class TransactionEventHandler
{
private readonly QueryDbContext _queryDbContext;
public TransactionEventHandler(QueryDbContext queryDbContext)
{
_queryDbContext = queryDbContext;
}
public async Task HandleAsync(TransactionEvent transactionEvent)
{
var transaction = new Transaction
{
Id = transactionEvent.Id,
Amount = transactionEvent.Amount,
Timestamp = transactionEvent.Timestamp,
UserId = transactionEvent.UserId,
Location = transactionEvent.Location
};
_queryDbContext.Transactions.Add(transaction);
await _queryDbContext.SaveChangesAsync();
}
}
2. Change Data Capture (CDC)
CDC is a technique used to track changes in the command database and propagate them to the query database. This can be done using database triggers or built-in CDC features provided by some databases.
Example
- Enable CDC: Enable CDC on the command database tables.
- Capture Changes: Use a service to capture changes and apply them to the query database.
EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',
@source_name = N'Transactions',
@role_name = NULL;
3. Transactional Outbox
The transactional outbox pattern ensures that events are reliably published whenever a transaction is committed. The outbox table stores events that need to be processed and published to the query database.
Example
- Outbox Table: Create an outbox table in the command database.
- Publish Events: A background service reads from the outbox table and updates the query database.
public class OutboxPublisherService
{
private readonly CommandDbContext _commandDbContext;
private readonly QueryDbContext _queryDbContext;
public OutboxPublisherService(CommandDbContext commandDbContext, QueryDbContext queryDbContext)
{
_commandDbContext = commandDbContext;
_queryDbContext = queryDbContext;
}
public async Task PublishEventsAsync()
{
var events = await _commandDbContext.OutboxEvents.ToListAsync();
foreach (var @event in events)
{
var transaction = new Transaction
{
Id = @event.TransactionId,
Amount = @event.Amount,
Timestamp = @event.Timestamp,
UserId = @event.UserId,
Location = @event.Location
};
_queryDbContext.Transactions.Add(transaction);
_commandDbContext.OutboxEvents.Remove(@event);
}
await _queryDbContext.SaveChangesAsync();
await _commandDbContext.SaveChangesAsync();
}
}
4. Data Synchronization Tools
Use data synchronization tools like SQL Server Data Tools (SSDT), dbForge Data Compare, or custom scripts to synchronize data between the command and query databases.
Example
- Compare and Synchronize: Use tools to compare and synchronize data periodically.
dbforge.datacompare /source connection:"Data Source=CommandDb;Initial Catalog=CommandDb;User ID=user;Password=pass" /target connection:"Data Source=QueryDb;Initial Catalog=QueryDb;User ID=user;Password=pass" /sync
Conclusion
Each of these strategies has its own advantages and trade-offs. The choice of strategy depends on your specific requirements, such as consistency, latency, and complexity. Implementing these patterns will help ensure that your command and query databases remain synchronized, providing a reliable and efficient CQRS setup.