using System.Linq.Expressions;
using Elsa.Common.Contracts;
using Elsa.Common.Entities;
using Elsa.Extensions;
using Elsa.MongoDb.Extensions;
using Elsa.Tenants;
using JetBrains.Annotations;
using MongoDB.Driver;
using MongoDB.Driver.Linq;
namespace Elsa.MongoDb.Common;
///
/// A generic repository class around MongoDb for accessing documents.
///
/// The type of the document.
[PublicAPI]
public class MongoDbStore(IMongoCollection collection, ITenantResolver tenantResolver)
where TDocument : class
{
///
/// Returns a queryable collection of documents.
///
public IMongoCollection GetCollection() => collection;
///
/// Saves the document.
///
/// The document to save.
/// The cancellation token.
public async Task AddAsync(TDocument document, CancellationToken cancellationToken = default)
{
await ApplyTenantIdAsync(document, cancellationToken);
await collection.InsertOneAsync(document, new InsertOneOptions(), cancellationToken);
return document;
}
///
/// Saves a list of documents.
///
/// The documents to save.
/// The cancellation token.
public async Task AddManyAsync(IEnumerable documents, CancellationToken cancellationToken = default)
{
var documentsList = documents.ToList();
if (!documentsList.Any())
return;
await ApplyTenantIdAsync(documentsList, cancellationToken);
await collection.InsertManyAsync(documentsList, new InsertManyOptions(), cancellationToken);
}
///
/// Saves the document.
///
/// The document to save.
/// The cancellation token.
public async Task SaveAsync(TDocument document, CancellationToken cancellationToken = default)
{
await ApplyTenantIdAsync(document, cancellationToken);
return await collection.FindOneAndReplaceAsync(document.BuildIdFilter(), document, new FindOneAndReplaceOptions
{
ReturnDocument = ReturnDocument.After,
IsUpsert = true
}, cancellationToken);
}
///
/// Saves the document.
///
/// The document to save.
/// The selector to use.
/// The cancellation token.
public async Task SaveAsync(TDocument document, Expression> selector, CancellationToken cancellationToken = default)
{
await ApplyTenantIdAsync(document, cancellationToken);
return await collection.FindOneAndReplaceAsync(document.BuildExpression(selector), document, new FindOneAndReplaceOptions
{
ReturnDocument = ReturnDocument.After,
IsUpsert = true
}, cancellationToken);
}
///
/// Saves the specified documents.
///
/// The documents to save.
/// The cancellation token.
public async Task SaveManyAsync(IEnumerable documents, CancellationToken cancellationToken = default)
{
var documentsList = documents.ToList();
await ApplyTenantIdAsync(documentsList, cancellationToken);
var writes = new List>();
foreach (var document in documentsList)
{
var replacement = new ReplaceOneModel(document.BuildIdFilter(), document)
{
IsUpsert = true
};
writes.Add(replacement);
}
if (!writes.Any())
return;
await collection.BulkWriteAsync(writes, cancellationToken: cancellationToken);
}
///
/// Saves the specified documents.
///
/// The documents to save.
/// The primary key to use.
/// The cancellation token.
public async Task SaveManyAsync(IEnumerable documents, string primaryKey = nameof(Entity.Id), CancellationToken cancellationToken = default)
{
var documentsList = documents.ToList();
await ApplyTenantIdAsync(documentsList, cancellationToken);
var writes = new List>();
foreach (var document in documentsList)
{
var replacement = new ReplaceOneModel(document.BuildFilter(primaryKey), document)
{
IsUpsert = true
};
writes.Add(replacement);
}
if (!writes.Any())
return;
await collection.BulkWriteAsync(writes, cancellationToken: cancellationToken);
}
///
/// Finds the document matching the specified predicate
///
/// The predicate to use.
/// The cancellation token.
/// The document if found, otherwise null.
public async Task FindAsync(Expression> predicate, CancellationToken cancellationToken = default)
{
return await FindAsync(predicate, false, cancellationToken);
}
///
/// Finds the document matching the specified predicate
///
/// The predicate to use.
/// Whether to include results across tenants.
/// The cancellation token.
/// The document if found, otherwise null.
public async Task FindAsync(Expression> predicate, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await queryable.Where(predicate).FirstOrDefaultAsync(cancellationToken);
}
///
/// Finds a single document using a query
///
/// The query to use
/// The cancellation token
/// The document if found, otherwise null
public async Task FindAsync(Func, IMongoQueryable> query, CancellationToken cancellationToken = default)
{
return await FindAsync(query, false, cancellationToken);
}
///
/// Finds a single document using a query
///
/// The query to use
/// Whether to include results across tenants
/// The cancellation token
/// The document if found, otherwise null
public async Task FindAsync(Func, IMongoQueryable> query, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await query(queryable).FirstOrDefaultAsync(cancellationToken);
}
///
/// Finds a list of documents matching the specified predicate
///
public async Task> FindManyAsync(Expression> predicate, CancellationToken cancellationToken = default)
{
return await FindManyAsync(predicate, false, cancellationToken);
}
///
/// Finds a list of documents matching the specified predicate
///
public async Task> FindManyAsync(Expression> predicate, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await queryable.Where(predicate).ToListAsync(cancellationToken);
}
///
/// Queries the database using a query and a selector.
///
public async Task> FindManyAsync(Func, IMongoQueryable> query, Expression> selector, CancellationToken cancellationToken = default)
{
return await FindManyAsync(query, selector, false, cancellationToken);
}
///
/// Queries the database using a query and a selector.
///
public async Task> FindManyAsync(Func, IMongoQueryable> query, Expression> selector, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await query(queryable).Select(selector).ToListAsync(cancellationToken);
}
///
/// Finds a list of documents using a query
///
public async Task> FindManyAsync(Func, IMongoQueryable> query, CancellationToken cancellationToken = default)
{
return await FindManyAsync(query, false, cancellationToken);
}
///
/// Finds a list of documents using a query
///
public async Task> FindManyAsync(Func, IMongoQueryable> query, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await query(queryable).ToListAsync(cancellationToken);
}
///
/// Queries the database using a query and a selector.
///
public async Task> FindMany(Func, IMongoQueryable> query, Expression> selector, CancellationToken cancellationToken = default)
{
return await FindMany(query, selector, false, cancellationToken);
}
///
/// Queries the database using a query and a selector.
///
public async Task> FindMany(Func, IMongoQueryable> query, Expression> selector, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await query(queryable).Select(selector).ToListAsync(cancellationToken);
}
///
/// Counts documents in the collection using a filter.
///
public async Task CountAsync(Func, IMongoQueryable> query, CancellationToken cancellationToken = default)
{
return await CountAsync(query, false, cancellationToken);
}
///
/// Counts documents in the collection using a filter.
///
public async Task CountAsync(Func, IMongoQueryable> query, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await query(queryable).LongCountAsync(cancellationToken);
}
///
/// Counts documents in the collection using a filter and distinct by a key selector.
///
public async Task CountAsync(Func, IMongoQueryable> query, Expression> propertySelector, CancellationToken cancellationToken = default)
{
return await CountAsync(query, propertySelector, false, cancellationToken);
}
///
/// Counts documents in the collection using a filter and distinct by a key selector.
///
public async Task CountAsync(Func, IMongoQueryable> query, Expression> propertySelector, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await query((IMongoQueryable)queryable.DistinctBy(propertySelector)).LongCountAsync(cancellationToken);
}
///
/// Checks if any documents exist.
///
public async Task AnyAsync(Expression> predicate, CancellationToken cancellationToken = default)
{
return await AnyAsync(predicate, false, cancellationToken);
}
///
/// Checks if any documents exist.
///
public async Task AnyAsync(Expression> predicate, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
return await queryable.Where(predicate).AnyAsync(cancellationToken);
}
///
/// Deletes documents using a predicate.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Expression> predicate, CancellationToken cancellationToken = default)
{
return await DeleteWhereAsync(predicate, false, cancellationToken);
}
///
/// Deletes documents using a predicate.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Expression> predicate, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
return await DeleteWhereAsync(predicate, nameof(Entity.Id), tenantAgnostic, cancellationToken);
}
///
/// Deletes documents using a predicate.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Expression> predicate, string key, CancellationToken cancellationToken = default)
{
return await DeleteWhereAsync(predicate, key, false, cancellationToken);
}
///
/// Deletes documents using a predicate.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Expression> predicate, string key, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
var documentsToDelete = await queryable.Where(predicate).ToListAsync(cancellationToken);
var count = documentsToDelete.LongCount();
var filter = documentsToDelete.BuildIdFilterForList(key);
await collection.DeleteManyAsync(filter, cancellationToken);
return count;
}
///
/// Deletes documents using a query.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Func, IMongoQueryable> query, Expression> keySelector, CancellationToken cancellationToken = default)
{
return await DeleteWhereAsync(query, keySelector, false, cancellationToken);
}
///
/// Deletes documents using a query.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Func, IMongoQueryable> query, Expression> keySelector, bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var key = keySelector.GetPropertyName();
return await DeleteWhereAsync(query, key, tenantAgnostic, cancellationToken);
}
///
/// Deletes documents using a query.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Func, IMongoQueryable> query, string key = nameof(Entity.Id), CancellationToken cancellationToken = default)
{
return await DeleteWhereAsync(query, key, false, cancellationToken);
}
///
/// Deletes documents using a query.
///
/// The number of documents deleted.
public async Task DeleteWhereAsync(Func, IMongoQueryable> query, string key = nameof(Entity.Id), bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = await GetQueryableCollectionAsync(tenantAgnostic, cancellationToken);
var documentsToDelete = await query(queryable).ToListAsync(cancellationToken);
var count = documentsToDelete.LongCount();
var filter = documentsToDelete.BuildIdFilterForList(key);
await collection.DeleteManyAsync(filter, cancellationToken);
return count;
}
private async Task> GetQueryableCollectionAsync(bool tenantAgnostic = false, CancellationToken cancellationToken = default)
{
var queryable = collection.AsQueryable();
if(tenantAgnostic)
return queryable;
if(typeof(Entity).IsAssignableFrom(typeof(TDocument)))
{
var tenant = await tenantResolver.GetTenantAsync(cancellationToken);
var tenantId = tenant?.Id;
queryable = queryable.Where(x => (x as Entity)!.TenantId == tenantId);
}
return queryable;
}
private async Task ApplyTenantIdAsync(TDocument document, CancellationToken cancellationToken)
{
var tenant = await tenantResolver.GetTenantAsync(cancellationToken);
var tenantId = tenant?.Id;
if (document is Entity tenantDocument)
tenantDocument.TenantId = tenantId;
}
private async Task ApplyTenantIdAsync(IEnumerable documents, CancellationToken cancellationToken)
{
var tenant = await tenantResolver.GetTenantAsync(cancellationToken);
var tenantId = tenant?.Id;
foreach (var document in documents)
{
if (document is Entity tenantDocument)
tenantDocument.TenantId = tenantId;
}
}
}