mirror of
https://github.com/opengram-server/opengram.git
synced 2026-09-12 21:44:13 +03:00
Initial commit
This commit is contained in:
zavolo
@@ -0,0 +1,26 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using MongoDB.Driver;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public class
|
||||
DefaultEventStoreMongoDbContext(IConfiguration configuration) : IMongoDbContext
|
||||
{
|
||||
private IMongoDatabase? _database;
|
||||
|
||||
public IMongoDatabase GetDatabase()
|
||||
{
|
||||
if (_database == null)
|
||||
{
|
||||
var connectionString = configuration.GetConnectionString(GetConnectionStringName());
|
||||
var databaseName = configuration.GetValue<string>(GetKeyOfDatabaseNameInConfiguration());
|
||||
var client = new MongoClient(connectionString);
|
||||
_database = client.GetDatabase(databaseName);
|
||||
}
|
||||
|
||||
return _database;
|
||||
}
|
||||
|
||||
protected virtual string GetConnectionStringName() => "Default";
|
||||
protected virtual string GetKeyOfDatabaseNameInConfiguration() => "App:DatabaseName";
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using MongoDB.Driver;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public class
|
||||
DefaultReadModelMongoDbContext(IConfiguration configuration) : IMongoDbContext
|
||||
{
|
||||
private IMongoDatabase? _database;
|
||||
|
||||
public IMongoDatabase GetDatabase()
|
||||
{
|
||||
if (_database == null)
|
||||
{
|
||||
var connectionString = configuration.GetConnectionString(GetConnectionStringName());
|
||||
var databaseName = configuration.GetValue<string>(GetKeyOfDatabaseNameInConfiguration());
|
||||
|
||||
// Configure MongoClient with proper read settings to avoid stale data
|
||||
var settings = MongoClientSettings.FromConnectionString(connectionString);
|
||||
settings.ReadConcern = ReadConcern.Local; // Read latest data without waiting for replication
|
||||
settings.ReadPreference = ReadPreference.Primary; // Always read from primary node
|
||||
settings.RetryReads = true; // Retry failed reads
|
||||
|
||||
var client = new MongoClient(settings);
|
||||
_database = client.GetDatabase(databaseName);
|
||||
}
|
||||
|
||||
return _database;
|
||||
}
|
||||
|
||||
protected virtual string GetConnectionStringName() => "Default";
|
||||
protected virtual string GetKeyOfDatabaseNameInConfiguration() => "App:ReadModelDatabaseName";
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
using EventFlow;
|
||||
using EventFlow.Aggregates;
|
||||
using EventFlow.Core;
|
||||
using EventFlow.Extensions;
|
||||
using EventFlow.MongoDB.EventStore;
|
||||
using EventFlow.MongoDB.ReadStores;
|
||||
using EventFlow.ReadStores;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
using MongoDB.Bson;
|
||||
using MongoDB.Bson.Serialization;
|
||||
using MongoDB.Bson.Serialization.Serializers;
|
||||
using MongoDB.Driver;
|
||||
using MyTelegram.EventFlow.MongoDB.ReadStores;
|
||||
using MyTelegram.EventFlow.ReadStores;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB.Extensions;
|
||||
|
||||
public static class MyMongoDbOptionsExtensions
|
||||
{
|
||||
public static void AddEventStoreMongoDbContext<TMongoDbContext>(this IServiceCollection services) where TMongoDbContext : class, IMongoDbContext
|
||||
{
|
||||
BsonSerializer.TryRegisterSerializer(new GuidSerializer(GuidRepresentation.Standard));
|
||||
services.TryAddSingleton<TMongoDbContext>();
|
||||
|
||||
services.TryAddSingleton<IMongoDatabase>(f => f.GetRequiredService<TMongoDbContext>().GetDatabase());
|
||||
}
|
||||
|
||||
public static void AddEventStoreMongoDbContext(this IServiceCollection services)
|
||||
{
|
||||
services.AddEventStoreMongoDbContext<DefaultEventStoreMongoDbContext>();
|
||||
}
|
||||
|
||||
public static void AddReadModelMongoDbContext<TMongoDbContext>(this IServiceCollection services)
|
||||
where TMongoDbContext : class, IMongoDbContext
|
||||
{
|
||||
BsonSerializer.TryRegisterSerializer(new GuidSerializer(GuidRepresentation.Standard));
|
||||
services.TryAddSingleton<TMongoDbContext>();
|
||||
services.TryAddSingleton<IMongoDbContext, TMongoDbContext>();
|
||||
services.TryAddSingleton<IQueryOnlyReadModelDescriptionProvider, QueryOnlyReadModelDescriptionProvider>();
|
||||
|
||||
services.TryAddSingleton<IReadModelDescriptionProvider, ReadModelDescriptionProvider>();
|
||||
services.TryAddSingleton<IMongoDbEventSequenceStore, MongoDbEventSequenceStore>();
|
||||
}
|
||||
|
||||
public static void AddReadModelMongoDbContext(this IServiceCollection services)
|
||||
{
|
||||
services.AddReadModelMongoDbContext<DefaultReadModelMongoDbContext>();
|
||||
}
|
||||
|
||||
public static IEventFlowOptions UseMongoDbReadModel<TAggregate, TIdentity, TReadModel>(
|
||||
this IEventFlowOptions eventFlowOptions)
|
||||
where TReadModel : class, IMongoDbReadModel
|
||||
where TIdentity : IIdentity
|
||||
where TAggregate : IAggregateRoot<TIdentity>
|
||||
{
|
||||
eventFlowOptions.ServiceCollection
|
||||
//.AddTransient<IMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel>>()
|
||||
.AddTransient<IMyMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel>>()
|
||||
.AddTransient<IQueryOnlyReadModelStore<TReadModel>, MongoDbQueryOnlyReadModelStore<TReadModel>>()
|
||||
;
|
||||
eventFlowOptions.ServiceCollection.AddTransient<IReadModelStore<TReadModel>>(f =>
|
||||
f.GetRequiredService<IMyMongoDbReadModelStore<TReadModel>>());
|
||||
#pragma warning disable CS0618
|
||||
eventFlowOptions.UseReadStoreFor<TAggregate, TIdentity, IMyMongoDbReadModelStore<TReadModel>, TReadModel>();
|
||||
#pragma warning restore CS0618
|
||||
|
||||
return eventFlowOptions;
|
||||
}
|
||||
|
||||
public static IEventFlowOptions UseMongoDbReadModel<TReadModel, TReadModelLocator>(
|
||||
this IEventFlowOptions eventFlowOptions)
|
||||
where TReadModel : class, IMongoDbReadModel
|
||||
where TReadModelLocator : IReadModelLocator
|
||||
{
|
||||
eventFlowOptions.ServiceCollection
|
||||
//.AddTransient<IMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel>>()
|
||||
.AddTransient<IMyMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel>>()
|
||||
.AddTransient<IQueryOnlyReadModelStore<TReadModel>, MongoDbQueryOnlyReadModelStore<TReadModel>>()
|
||||
;
|
||||
|
||||
eventFlowOptions.ServiceCollection.AddTransient<IReadModelStore<TReadModel>>(f =>
|
||||
f.GetRequiredService<IMyMongoDbReadModelStore<TReadModel>>());
|
||||
eventFlowOptions.UseReadStoreFor<IMyMongoDbReadModelStore<TReadModel>, TReadModel, TReadModelLocator>();
|
||||
|
||||
return eventFlowOptions;
|
||||
}
|
||||
|
||||
private static void AddMongoDbStoreServices<TReadModel, TDbContext>(this IServiceCollection services)
|
||||
where TReadModel : class, IQueryOnlyReadModel
|
||||
where TDbContext : IMongoDbContext
|
||||
{
|
||||
//services.AddTransient<IMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel, TDbContext>>()
|
||||
services
|
||||
.AddTransient<IQueryOnlyReadModelStore<TReadModel>, MongoDbQueryOnlyReadModelStore<TReadModel, TDbContext>>();
|
||||
}
|
||||
|
||||
public static IEventFlowOptions UseMongoDbReadModel<TAggregate, TIdentity, TReadModel, TDbContext>(
|
||||
this IEventFlowOptions eventFlowOptions)
|
||||
where TReadModel : class, IMongoDbReadModel
|
||||
where TIdentity : IIdentity
|
||||
where TAggregate : IAggregateRoot<TIdentity>
|
||||
where TDbContext : class, IMongoDbContext
|
||||
{
|
||||
eventFlowOptions.ServiceCollection
|
||||
.AddTransient<IMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel, TDbContext>>()
|
||||
.AddTransient<IMyMongoDbReadModelStore<TReadModel>,
|
||||
MyMongoDbReadModelStore<TReadModel, TDbContext>>()
|
||||
.AddTransient<IMyMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel, TDbContext>>()
|
||||
.AddTransient<IQueryOnlyReadModelStore<TReadModel>, MongoDbQueryOnlyReadModelStore<TReadModel, TDbContext>>()
|
||||
;
|
||||
eventFlowOptions.ServiceCollection.AddTransient<IReadModelStore<TReadModel>>(f =>
|
||||
f.GetRequiredService<IMyMongoDbReadModelStore<TReadModel>>());
|
||||
//eventFlowOptions.UseReadStoreFor<IMongoDbReadModelStore<TReadModel>, TReadModel>();
|
||||
#pragma warning disable CS0618
|
||||
eventFlowOptions
|
||||
.UseReadStoreFor<TAggregate, TIdentity, IMongoDbReadModelStore<TReadModel>, TReadModel>();
|
||||
#pragma warning restore CS0618
|
||||
|
||||
return eventFlowOptions;
|
||||
}
|
||||
|
||||
public static IEventFlowOptions UseMongoDbReadModelWithContext<TReadModel, TMongoDbContext>(
|
||||
this IEventFlowOptions eventFlowOptions)
|
||||
where TReadModel : class, IMongoDbReadModel
|
||||
where TMongoDbContext : class, IMongoDbContext
|
||||
{
|
||||
eventFlowOptions.ServiceCollection
|
||||
.AddTransient<IMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel>>()
|
||||
.AddTransient<IMyMongoDbReadModelStore<TReadModel>,
|
||||
MyMongoDbReadModelStore<TReadModel, TMongoDbContext>>()
|
||||
.AddTransient<IReadModelStore<TReadModel>>(f =>
|
||||
f.GetRequiredService<IMyMongoDbReadModelStore<TReadModel>>())
|
||||
;
|
||||
|
||||
eventFlowOptions.UseReadStoreFor<IMongoDbReadModelStore<TReadModel>, TReadModel>();
|
||||
|
||||
return eventFlowOptions;
|
||||
}
|
||||
|
||||
public static IEventFlowOptions UseMongoDbReadModelWithContext<TReadModel, TReadModelLocator, TMongoDbContext>(
|
||||
this IEventFlowOptions eventFlowOptions)
|
||||
where TReadModel : class, IMongoDbReadModel
|
||||
where TMongoDbContext : class, IMongoDbContext
|
||||
where TReadModelLocator : IReadModelLocator
|
||||
{
|
||||
eventFlowOptions.ServiceCollection
|
||||
.AddTransient<IMongoDbReadModelStore<TReadModel>, MyMongoDbReadModelStore<TReadModel>>()
|
||||
.AddTransient<IMyMongoDbReadModelStore<TReadModel>,
|
||||
MyMongoDbReadModelStore<TReadModel, TMongoDbContext>>()
|
||||
.AddTransient<IReadModelStore<TReadModel>>(f =>
|
||||
f.GetRequiredService<IMyMongoDbReadModelStore<TReadModel>>())
|
||||
;
|
||||
|
||||
eventFlowOptions.UseReadStoreFor<IMongoDbReadModelStore<TReadModel>, TReadModel, TReadModelLocator>();
|
||||
|
||||
return eventFlowOptions;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
|
||||
<ConfigureAwait />
|
||||
</Weavers>
|
||||
@@ -0,0 +1,30 @@
|
||||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
|
||||
<!-- This file was generated by Fody. Manual changes to this file will be lost when your project is rebuilt. -->
|
||||
<xs:element name="Weavers">
|
||||
<xs:complexType>
|
||||
<xs:all>
|
||||
<xs:element name="ConfigureAwait" minOccurs="0" maxOccurs="1">
|
||||
<xs:complexType>
|
||||
<xs:attribute name="ContinueOnCapturedContext" type="xs:boolean" />
|
||||
</xs:complexType>
|
||||
</xs:element>
|
||||
</xs:all>
|
||||
<xs:attribute name="VerifyAssembly" type="xs:boolean">
|
||||
<xs:annotation>
|
||||
<xs:documentation>'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed.</xs:documentation>
|
||||
</xs:annotation>
|
||||
</xs:attribute>
|
||||
<xs:attribute name="VerifyIgnoreCodes" type="xs:string">
|
||||
<xs:annotation>
|
||||
<xs:documentation>A comma-separated list of error codes that can be safely ignored in assembly verification.</xs:documentation>
|
||||
</xs:annotation>
|
||||
</xs:attribute>
|
||||
<xs:attribute name="GenerateXsd" type="xs:boolean">
|
||||
<xs:annotation>
|
||||
<xs:documentation>'false' to turn off automatic generation of the XML Schema file.</xs:documentation>
|
||||
</xs:annotation>
|
||||
</xs:attribute>
|
||||
</xs:complexType>
|
||||
</xs:element>
|
||||
</xs:schema>
|
||||
@@ -0,0 +1,8 @@
|
||||
using MongoDB.Driver;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public interface IMongoDbContext
|
||||
{
|
||||
IMongoDatabase GetDatabase();
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public interface IMongoDbIndexesCreator
|
||||
{
|
||||
Task CreateAllIndexesAsync();
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
using System.Linq.Expressions;
|
||||
using EventFlow.MongoDB.ReadStores;
|
||||
using EventFlow.ReadStores;
|
||||
using MongoDB.Driver;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public interface IMyMongoDbReadModelStore<TReadModel> : IMongoDbReadModelStore<TReadModel> where TReadModel : class, IReadModel
|
||||
{
|
||||
Task<IAsyncCursor<TResult>> FindAsync<TResult>(
|
||||
Expression<Func<TReadModel, bool>> filter,
|
||||
FindOptions<TReadModel, TResult>? options = null,
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
Task<long> CountAsync(Expression<Func<TReadModel, bool>>? filter = null, CancellationToken cancellationToken = default);
|
||||
|
||||
Task<IAggregateFluent<TResult>> AggregateAsync<TResult, TKey>(
|
||||
Expression<Func<TReadModel, bool>> filter,
|
||||
Expression<Func<TReadModel, TKey>> id,
|
||||
Expression<Func<IGrouping<TKey, TReadModel>, TResult>> group,
|
||||
AggregateOptions? options = null,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
using MongoDB.Driver;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public class MongoDbContext(IMongoDatabase mongoDatabase) : IMongoDbContext
|
||||
{
|
||||
public IMongoDatabase GetDatabase() => mongoDatabase;
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
using EventFlow.MongoDB.EventStore;
|
||||
using EventFlow.MongoDB.ReadStores;
|
||||
using EventFlow.MongoDB.ValueObjects;
|
||||
using MongoDB.Driver;
|
||||
using System.Linq.Expressions;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public abstract class MongoDbIndexesCreatorBase(
|
||||
IMongoDatabase database,
|
||||
IReadModelDescriptionProvider descriptionProvider,
|
||||
IMongoDbEventPersistenceInitializer eventPersistenceInitializer)
|
||||
: IMongoDbIndexesCreator
|
||||
{
|
||||
public async Task CreateAllIndexesAsync()
|
||||
{
|
||||
eventPersistenceInitializer.Initialize();
|
||||
var snapShotCollectionName = "snapShots";
|
||||
await CreateIndexAsync<MongoDbSnapshotDataModel>(p => p.AggregateId, snapShotCollectionName);
|
||||
await CreateIndexAsync<MongoDbSnapshotDataModel>(p => p.AggregateName, snapShotCollectionName);
|
||||
await CreateIndexAsync<MongoDbSnapshotDataModel>(p => p.AggregateSequenceNumber, snapShotCollectionName);
|
||||
|
||||
await CreateAllIndexesCoreAsync();
|
||||
}
|
||||
|
||||
protected abstract Task CreateAllIndexesCoreAsync();
|
||||
|
||||
protected async Task CreateIndexAsync<TReadModel>(Expression<Func<TReadModel, object>> field)
|
||||
where TReadModel : IMongoDbReadModel
|
||||
{
|
||||
var indexDefine = Builders<TReadModel>.IndexKeys.Ascending(field);
|
||||
var collectionName = descriptionProvider.GetReadModelDescription<TReadModel>().RootCollectionName;
|
||||
await database.GetCollection<TReadModel>(collectionName.Value).Indexes
|
||||
.CreateOneAsync(new CreateIndexModel<TReadModel>(indexDefine));
|
||||
}
|
||||
|
||||
protected async Task CreateIndexAsync<TSnapshot>(Expression<Func<TSnapshot, object>> field,
|
||||
string collectionName)
|
||||
{
|
||||
var indexDefine = Builders<TSnapshot>.IndexKeys.Ascending(field);
|
||||
await database.GetCollection<TSnapshot>(collectionName).Indexes
|
||||
.CreateOneAsync(new CreateIndexModel<TSnapshot>(indexDefine));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,337 @@
|
||||
using EventFlow.Aggregates;
|
||||
using EventFlow.Core;
|
||||
using EventFlow.Core.RetryStrategies;
|
||||
using EventFlow.Extensions;
|
||||
using EventFlow.MongoDB.ReadStores;
|
||||
using EventFlow.ReadStores;
|
||||
using EventFlow.Exceptions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MongoDB.Driver;
|
||||
using System.Linq.Expressions;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB;
|
||||
|
||||
public class MyMongoDbReadModelStore<TReadModel>(
|
||||
ILogger<MongoDbReadModelStore<TReadModel>> logger,
|
||||
IReadModelDescriptionProvider readModelDescriptionProvider,
|
||||
ITransientFaultHandler<IOptimisticConcurrencyRetryStrategy> transientFaultHandler,
|
||||
IMongoDbContext mongoDbContext)
|
||||
:
|
||||
MyMongoDbReadModelStore<TReadModel, IMongoDbContext>(logger, readModelDescriptionProvider,
|
||||
transientFaultHandler, mongoDbContext)
|
||||
where TReadModel : class, IMongoDbReadModel;
|
||||
|
||||
public class MyMongoDbReadModelStore<TReadModel, TDbContext> : IMyMongoDbReadModelStore<TReadModel>
|
||||
where TReadModel : class, IMongoDbReadModel
|
||||
where TDbContext : IMongoDbContext
|
||||
{
|
||||
private readonly ILogger<MongoDbReadModelStore<TReadModel>> _logger;
|
||||
private readonly IReadModelDescriptionProvider _readModelDescriptionProvider;
|
||||
private readonly ITransientFaultHandler<IOptimisticConcurrencyRetryStrategy> _transientFaultHandler;
|
||||
private readonly TDbContext _dbContext;
|
||||
private readonly IMongoDatabase _database;
|
||||
|
||||
public MyMongoDbReadModelStore(
|
||||
ILogger<MongoDbReadModelStore<TReadModel>> logger,
|
||||
IReadModelDescriptionProvider readModelDescriptionProvider,
|
||||
ITransientFaultHandler<IOptimisticConcurrencyRetryStrategy> transientFaultHandler,
|
||||
TDbContext dbContext)
|
||||
{
|
||||
_logger = logger;
|
||||
_readModelDescriptionProvider = readModelDescriptionProvider;
|
||||
_transientFaultHandler = transientFaultHandler;
|
||||
_dbContext = dbContext;
|
||||
_database = dbContext.GetDatabase();
|
||||
}
|
||||
|
||||
private IMongoDatabase GetDatabase() => _database;
|
||||
|
||||
public async Task UpdateAsync(
|
||||
IReadOnlyCollection<ReadModelUpdate> readModelUpdates,
|
||||
IReadModelContextFactory readModelContextFactory,
|
||||
Func<IReadModelContext, IReadOnlyCollection<IDomainEvent>, ReadModelEnvelope<TReadModel>, CancellationToken, Task<ReadModelUpdateResult<TReadModel>>> updateReadModel,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogTrace(
|
||||
"UpdateAsync called with {Count} read model updates for type '{ReadModel}'",
|
||||
readModelUpdates.Count,
|
||||
typeof(TReadModel).PrettyPrint());
|
||||
|
||||
foreach (var readModelUpdate in readModelUpdates)
|
||||
{
|
||||
_logger.LogTrace(
|
||||
"Processing read model update for '{ReadModelId}' with {EventCount} domain events",
|
||||
readModelUpdate.ReadModelId,
|
||||
readModelUpdate.DomainEvents.Count);
|
||||
|
||||
// Retry logic with reload - each retry will fetch fresh data from MongoDB
|
||||
await _transientFaultHandler.TryAsync(
|
||||
async c =>
|
||||
{
|
||||
await UpdateReadModelAsync(
|
||||
readModelUpdate.ReadModelId,
|
||||
readModelUpdate,
|
||||
readModelContextFactory,
|
||||
updateReadModel,
|
||||
c);
|
||||
return 0; // Success
|
||||
},
|
||||
Label.Named("mongodb-read-model-update"),
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task UpdateReadModelAsync(
|
||||
string readModelId,
|
||||
ReadModelUpdate readModelUpdate,
|
||||
IReadModelContextFactory readModelContextFactory,
|
||||
Func<IReadModelContext, IReadOnlyCollection<IDomainEvent>, ReadModelEnvelope<TReadModel>, CancellationToken, Task<ReadModelUpdateResult<TReadModel>>> updateReadModel,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
try
|
||||
{
|
||||
// Try to get existing read model - ALWAYS reload from database to get fresh version
|
||||
var filter = Builders<TReadModel>.Filter.Eq("_id", readModelId);
|
||||
var existingReadModel = await collection.Find(filter).FirstOrDefaultAsync(cancellationToken);
|
||||
|
||||
_logger.LogTrace(
|
||||
"Loaded read model '{ReadModelId}' from MongoDB - Exists: {Exists}, Version: {Version}",
|
||||
readModelId,
|
||||
existingReadModel != null,
|
||||
existingReadModel?.Version);
|
||||
|
||||
ReadModelEnvelope<TReadModel> readModelEnvelope;
|
||||
long? originalVersion = existingReadModel?.Version; // Save original version BEFORE modification
|
||||
|
||||
if (existingReadModel == null)
|
||||
{
|
||||
// Create new read model
|
||||
_logger.LogTrace(
|
||||
"Creating new read model '{ReadModelId}' of type '{ReadModel}'",
|
||||
readModelId,
|
||||
typeof(TReadModel).PrettyPrint());
|
||||
|
||||
readModelEnvelope = ReadModelEnvelope<TReadModel>.With(readModelId, default(TReadModel));
|
||||
}
|
||||
else
|
||||
{
|
||||
// Update existing read model
|
||||
_logger.LogTrace(
|
||||
"Updating existing read model '{ReadModelId}' of type '{ReadModel}' with version '{Version}'",
|
||||
readModelId,
|
||||
typeof(TReadModel).PrettyPrint(),
|
||||
existingReadModel.Version);
|
||||
|
||||
readModelEnvelope = ReadModelEnvelope<TReadModel>.With(readModelId, existingReadModel);
|
||||
}
|
||||
|
||||
_logger.LogTrace(
|
||||
"Before applying events - ReadModel '{ReadModelId}' envelope Version: '{EnvelopeVersion}', IsNew: {IsNew}",
|
||||
readModelId,
|
||||
readModelEnvelope.ReadModel?.Version,
|
||||
existingReadModel == null);
|
||||
|
||||
var readModelContext = readModelContextFactory.Create(readModelId, isNew: existingReadModel == null);
|
||||
var result = await updateReadModel(readModelContext, readModelUpdate.DomainEvents, readModelEnvelope, cancellationToken);
|
||||
|
||||
if (!result.IsModified)
|
||||
{
|
||||
_logger.LogTrace(
|
||||
"Read model '{ReadModelId}' was not modified",
|
||||
readModelId);
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogTrace(
|
||||
"After applying {EventCount} events: ReadModel '{ReadModelId}' new version will be '{NewVersion}', was '{OldVersion}'",
|
||||
readModelUpdate.DomainEvents.Count,
|
||||
readModelId,
|
||||
result.Envelope.ReadModel?.Version,
|
||||
readModelEnvelope.ReadModel?.Version);
|
||||
|
||||
// Save the updated read model
|
||||
if (existingReadModel == null)
|
||||
{
|
||||
// Insert new document
|
||||
try
|
||||
{
|
||||
await collection.InsertOneAsync(result.Envelope.ReadModel, cancellationToken: cancellationToken);
|
||||
|
||||
_logger.LogTrace(
|
||||
"Inserted new read model '{ReadModelId}' with version '{Version}'",
|
||||
readModelId,
|
||||
result.Envelope.ReadModel.Version);
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
// Another process inserted it concurrently, throw exception to retry with reload
|
||||
_logger.LogWarning(
|
||||
ex,
|
||||
"Duplicate key error when inserting read model '{ReadModelId}'. Another process created it concurrently. Throwing OptimisticConcurrencyException for retry.",
|
||||
readModelId);
|
||||
|
||||
// Add a small delay to reduce contention on retry
|
||||
await Task.Delay(50, cancellationToken);
|
||||
|
||||
throw new OptimisticConcurrencyException($"Read model '{readModelId}' created by another process", ex);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Update existing document with optimistic concurrency check
|
||||
_logger.LogTrace(
|
||||
"Preparing to update - originalVersion: {OriginalVersion}, existingReadModel.Version (after Apply): {ExistingVersion}, result.Envelope.ReadModel.Version: {ResultVersion}, Are they same object: {SameObject}",
|
||||
originalVersion,
|
||||
existingReadModel.Version,
|
||||
result.Envelope.ReadModel.Version,
|
||||
ReferenceEquals(existingReadModel, result.Envelope.ReadModel));
|
||||
|
||||
// В фильтр берём originalVersion (зафиксированную до вызова Apply), а не existingReadModel.Version, которую Apply уже изменил
|
||||
var updateFilter = Builders<TReadModel>.Filter.And(
|
||||
Builders<TReadModel>.Filter.Eq("_id", readModelId),
|
||||
Builders<TReadModel>.Filter.Eq(r => r.Version, originalVersion));
|
||||
|
||||
var replaceResult = await collection.ReplaceOneAsync(
|
||||
updateFilter,
|
||||
result.Envelope.ReadModel,
|
||||
new ReplaceOptions { IsUpsert = false },
|
||||
cancellationToken);
|
||||
|
||||
if (replaceResult.MatchedCount == 0)
|
||||
{
|
||||
// Document was modified by another process - reload and retry
|
||||
var currentInDb = await collection.Find(Builders<TReadModel>.Filter.Eq("_id", readModelId)).FirstOrDefaultAsync(cancellationToken);
|
||||
|
||||
_logger.LogWarning(
|
||||
"Optimistic concurrency conflict for read model '{ReadModelId}'. We loaded version '{LoadedVersion}' and tried to save version '{NewVersion}', but database now has version '{CurrentVersion}'. Throwing OptimisticConcurrencyException to reload and retry. Events count: {EventsCount}",
|
||||
readModelId,
|
||||
originalVersion, // Use the original version we tried to match
|
||||
result.Envelope.ReadModel.Version,
|
||||
currentInDb?.Version,
|
||||
readModelUpdate.DomainEvents.Count);
|
||||
|
||||
// Add a small delay to reduce contention on retry
|
||||
await Task.Delay(50, cancellationToken);
|
||||
|
||||
throw new OptimisticConcurrencyException($"Read model '{readModelId}' updated by another process");
|
||||
}
|
||||
|
||||
_logger.LogTrace(
|
||||
"Updated read model '{ReadModelId}' from version '{OldVersion}' to '{NewVersion}'",
|
||||
readModelId,
|
||||
originalVersion,
|
||||
result.Envelope.ReadModel.Version);
|
||||
}
|
||||
}
|
||||
catch (MongoWriteException ex) when (ex.WriteError?.Category == ServerErrorCategory.DuplicateKey)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
ex,
|
||||
"Duplicate key error for read model '{ReadModelId}'. Another process likely modified it concurrently. Throwing OptimisticConcurrencyException for retry.",
|
||||
readModelId);
|
||||
|
||||
throw new OptimisticConcurrencyException($"Read model '{readModelId}' updated by another process", ex);
|
||||
}
|
||||
}
|
||||
|
||||
public Task DeleteAsync(string id, CancellationToken cancellationToken)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
var filter = Builders<TReadModel>.Filter.Eq("_id", id);
|
||||
return collection.DeleteOneAsync(filter, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<ReadModelEnvelope<TReadModel>> GetAsync(string id, CancellationToken cancellationToken)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
var filter = Builders<TReadModel>.Filter.Eq("_id", id);
|
||||
var readModel = await collection.Find(filter).FirstOrDefaultAsync(cancellationToken);
|
||||
|
||||
return readModel == null
|
||||
? ReadModelEnvelope<TReadModel>.Empty(id)
|
||||
: ReadModelEnvelope<TReadModel>.With(id, readModel);
|
||||
}
|
||||
|
||||
public Task<IAggregateFluent<TResult>> AggregateAsync<TResult, TKey>(
|
||||
Expression<Func<TReadModel, bool>> filter,
|
||||
Expression<Func<TReadModel, TKey>> id,
|
||||
Expression<Func<IGrouping<TKey, TReadModel>, TResult>> group,
|
||||
AggregateOptions? options = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
return Task.FromResult(collection.Aggregate()
|
||||
.Match(filter)
|
||||
.Group(id, group))
|
||||
;
|
||||
}
|
||||
|
||||
public async Task<IAsyncCursor<TResult>> FindAsync<TResult>(Expression<Func<TReadModel, bool>> filter, FindOptions<TReadModel, TResult>? options = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
_logger.LogTrace(
|
||||
"Finding read model '{ReadModel}' with expression '{Filter}' from collection '{RootCollectionName}'",
|
||||
typeof(TReadModel).PrettyPrint(),
|
||||
filter,
|
||||
readModelDescription.RootCollectionName);
|
||||
|
||||
return await collection.FindAsync(filter, options, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<long> CountAsync(Expression<Func<TReadModel, bool>>? filter = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
return collection.CountDocumentsAsync(filter, cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<IAsyncCursor<TReadModel>> FindAsync(
|
||||
Expression<Func<TReadModel, bool>> filter,
|
||||
FindOptions<TReadModel, TReadModel>? options = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
_logger.LogTrace(
|
||||
"Finding read model '{ReadModel}' with expression '{Filter}' from collection '{RootCollectionName}'",
|
||||
typeof(TReadModel).PrettyPrint(),
|
||||
filter,
|
||||
readModelDescription.RootCollectionName);
|
||||
|
||||
return await collection.FindAsync(filter, options, cancellationToken);
|
||||
}
|
||||
|
||||
public IQueryable<TReadModel> AsQueryable()
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
return collection.AsQueryable();
|
||||
}
|
||||
|
||||
public Task DeleteAllAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var readModelDescription = _readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
_logger.LogWarning(
|
||||
"Deleting all documents from collection '{CollectionName}'",
|
||||
readModelDescription.RootCollectionName);
|
||||
|
||||
return collection.DeleteManyAsync(Builders<TReadModel>.Filter.Empty, cancellationToken);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net9.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<None Include="FodyWeavers.xsd" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="EventFlow.MongoDB" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Console" />
|
||||
<ProjectReference Include="..\MyTelegram.EventFlow\MyTelegram.EventFlow.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Update="Fody">
|
||||
<PrivateAssets>all</PrivateAssets>
|
||||
<IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
|
||||
</PackageReference>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
using EventFlow.MongoDB.ValueObjects;
|
||||
using EventFlow.ReadStores;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB.ReadStores;
|
||||
|
||||
public interface IQueryOnlyReadModelDescriptionProvider
|
||||
{
|
||||
ReadModelDescription GetReadModelDescription<TReadModel>()
|
||||
where TReadModel : IReadModel;
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
using EventFlow.Extensions;
|
||||
using EventFlow.ReadStores;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MongoDB.Driver;
|
||||
using MyTelegram.EventFlow.ReadStores;
|
||||
using System.Linq.Expressions;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB.ReadStores;
|
||||
|
||||
public class MongoDbQueryOnlyReadModelStore<TReadModel>(
|
||||
IQueryOnlyReadModelDescriptionProvider readModelDescriptionProvider,
|
||||
IMongoDbContext dbContext,
|
||||
ILogger<MongoDbQueryOnlyReadModelStore<TReadModel, IMongoDbContext>> logger)
|
||||
: MongoDbQueryOnlyReadModelStore<TReadModel, IMongoDbContext>(readModelDescriptionProvider, dbContext,
|
||||
logger)
|
||||
where TReadModel : class, IReadModel;
|
||||
|
||||
public class MongoDbQueryOnlyReadModelStore<TReadModel, TDbContext>(
|
||||
IQueryOnlyReadModelDescriptionProvider readModelDescriptionProvider,
|
||||
TDbContext dbContext,
|
||||
ILogger<MongoDbQueryOnlyReadModelStore<TReadModel, TDbContext>> logger)
|
||||
: IQueryOnlyReadModelStore<TReadModel>
|
||||
where TReadModel : class, IReadModel
|
||||
where TDbContext : IMongoDbContext
|
||||
{
|
||||
public IQueryable<TReadModel> GetAll()
|
||||
{
|
||||
var readModelDescription = readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
return collection.AsQueryable();
|
||||
}
|
||||
|
||||
public Task<IReadOnlyCollection<TReadModel>> FindAsync(Expression<Func<TReadModel, bool>> filter, int skip = 0,
|
||||
int limit = 0,
|
||||
SortOptions<TReadModel>? sort = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
return FindAsync(filter, p => p, skip, limit, sort, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<IReadOnlyCollection<TResult>> FindAsync<TResult>(Expression<Func<TReadModel, bool>> filter,
|
||||
Expression<Func<TReadModel, TResult>> createResult, int skip = 0,
|
||||
int limit = 0, SortOptions<TReadModel>? sort = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var findOptions = CreateFindOptions(createResult, skip, limit, sort);
|
||||
var cursor = await FindCoreAsync(filter, findOptions, cancellationToken);
|
||||
|
||||
return await cursor.ToListAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<TReadModel?> FirstOrDefaultAsync(Expression<Func<TReadModel, bool>> filter,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var cursor = await FindCoreAsync<TReadModel>(filter, cancellationToken: cancellationToken);
|
||||
return await cursor.FirstOrDefaultAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<TResult?> FirstOrDefaultAsync<TResult>(Expression<Func<TReadModel, bool>> filter,
|
||||
Expression<Func<TReadModel, TResult>> createResult,
|
||||
SortOptions<TReadModel>? sort = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var findOptions = CreateFindOptions(createResult, 0, 0, sort);
|
||||
|
||||
var cursor = await FindCoreAsync(filter, findOptions, cancellationToken);
|
||||
return await cursor.FirstOrDefaultAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public Task<long> CountAsync(Expression<Func<TReadModel, bool>> filter,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var readModelDescription = readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
return collection.CountDocumentsAsync(filter, cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
private static FindOptions<T1, T2> CreateFindOptions<T1, T2>(
|
||||
Expression<Func<T1, T2>> createResult,
|
||||
int skip = 0,
|
||||
int limit = 0, SortOptions<T1>? sort = null)
|
||||
{
|
||||
var findOptions = new FindOptions<T1, T2>
|
||||
{
|
||||
Projection = Builders<T1>.Projection.Expression(createResult)
|
||||
};
|
||||
if (skip > 0)
|
||||
{
|
||||
findOptions.Skip = skip;
|
||||
}
|
||||
|
||||
if (limit > 0)
|
||||
{
|
||||
findOptions.Limit = limit;
|
||||
}
|
||||
|
||||
if (sort != null)
|
||||
{
|
||||
findOptions.Sort = sort.SortType switch
|
||||
{
|
||||
SortType.None or SortType.Ascending => Builders<T1>.Sort.Ascending(sort.Sort),
|
||||
SortType.Descending => Builders<T1>.Sort.Descending(sort.Sort),
|
||||
_ => throw new ArgumentOutOfRangeException()
|
||||
};
|
||||
}
|
||||
|
||||
return findOptions;
|
||||
}
|
||||
|
||||
private async Task<IAsyncCursor<TResult>> FindCoreAsync<TResult>(Expression<Func<TReadModel, bool>> filter,
|
||||
FindOptions<TReadModel, TResult>? options = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var readModelDescription = readModelDescriptionProvider.GetReadModelDescription<TReadModel>();
|
||||
var collection = GetDatabase().GetCollection<TReadModel>(readModelDescription.RootCollectionName.Value);
|
||||
|
||||
logger.LogTrace(
|
||||
"Finding read model '{ReadModel}' with expression '{Filter}' from collection '{RootCollectionName}'",
|
||||
typeof(TReadModel).PrettyPrint(),
|
||||
filter,
|
||||
readModelDescription.RootCollectionName);
|
||||
|
||||
return await collection.FindAsync(filter, options, cancellationToken);
|
||||
}
|
||||
|
||||
private IMongoDatabase GetDatabase()
|
||||
{
|
||||
return dbContext.GetDatabase();
|
||||
}
|
||||
}
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Reflection;
|
||||
using EventFlow.Extensions;
|
||||
using EventFlow.MongoDB.ReadStores.Attributes;
|
||||
using EventFlow.MongoDB.ValueObjects;
|
||||
using EventFlow.ReadStores;
|
||||
|
||||
namespace MyTelegram.EventFlow.MongoDB.ReadStores;
|
||||
|
||||
public class QueryOnlyReadModelDescriptionProvider : IQueryOnlyReadModelDescriptionProvider
|
||||
{
|
||||
private static readonly ConcurrentDictionary<Type, ReadModelDescription> CollectionNames = new();
|
||||
|
||||
public ReadModelDescription GetReadModelDescription<TReadModel>() where TReadModel : IReadModel
|
||||
{
|
||||
var name = typeof(TReadModel).PrettyPrint().ToLowerInvariant();
|
||||
if (typeof(TReadModel).IsInterface && name.StartsWith("i"))
|
||||
{
|
||||
name = name[1..];
|
||||
}
|
||||
|
||||
return CollectionNames.GetOrAdd(
|
||||
typeof(TReadModel),
|
||||
t =>
|
||||
{
|
||||
var collectionType = t.GetTypeInfo().GetCustomAttribute<MongoDbCollectionNameAttribute>();
|
||||
var indexName = collectionType == null
|
||||
? $"eventflow-{name}"
|
||||
: collectionType.CollectionName;
|
||||
return new ReadModelDescription(new RootCollectionName(indexName));
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user