--- url: /guide/what-is-foundatio-repositories.md --- # What is Foundatio.Repositories? Foundatio.Repositories is a production-grade repository pattern library for .NET that provides a clean abstraction over data access with powerful features like caching, messaging, soft deletes, and versioning. It's built on top of [Foundatio](https://github.com/FoundatioFx/Foundatio) building blocks and currently provides a full-featured Elasticsearch implementation. ## Why Use Foundatio.Repositories? Building robust data access layers requires handling many cross-cutting concerns: * **Caching** - How do you cache entities and invalidate them when they change? * **Notifications** - How do you notify other parts of your system when data changes? * **Soft Deletes** - How do you implement soft deletes consistently across all queries? * **Versioning** - How do you handle optimistic concurrency? * **Querying** - How do you build dynamic, user-facing queries safely? Foundatio.Repositories solves all of these problems with a cohesive, well-tested implementation. ## Architecture Overview ```mermaid flowchart TB subgraph Application App[Your Application] end subgraph Repository["Repository Layer"] IRepo["IRepository<T>"] ISearch["ISearchableRepository<T>"] IReadOnly["IReadOnlyRepository<T>"] end subgraph Elasticsearch["Elasticsearch Implementation"] ElasticRepo["ElasticRepositoryBase<T>"] Index["Index Configuration"] QueryBuilder["Query Builders"] end subgraph Foundatio["Foundatio Building Blocks"] Cache["ICacheClient"] MessageBus["IMessageBus"] Lock["ILockProvider"] Queue["IQueue"] end subgraph Storage["Data Storage"] ES[(Elasticsearch)] end App --> IRepo App --> ISearch App --> IReadOnly IRepo --> ElasticRepo ISearch --> ElasticRepo IReadOnly --> ElasticRepo ElasticRepo --> Index ElasticRepo --> QueryBuilder ElasticRepo --> Cache ElasticRepo --> MessageBus ElasticRepo --> ES ``` ## Key Features ### Repository Pattern Clean interfaces that abstract data access: * **`IReadOnlyRepository`** - Read operations (Get, Find, Count, Exists) * **`IRepository`** - Write operations (Add, Save, Remove, Patch) * **`ISearchableRepository`** - Dynamic querying with filters, sorting, and aggregations ### Elasticsearch Implementation Full-featured Elasticsearch support: * Index configuration with schema versioning * Daily and monthly index strategies for time-series data * Parent-child document relationships * Custom field mappings and analyzers ### Built on Foundatio Leverages Foundatio's battle-tested building blocks: * **Caching** - Distributed cache with automatic invalidation * **Messaging** - Entity change notifications via message bus * **Locking** - Distributed locks for coordination * **Queues** - Background job processing ### Developer Experience * Async/await throughout * Strongly-typed queries with lambda expressions * Comprehensive event system for extensibility * Detailed logging and diagnostics ## Use Cases Foundatio.Repositories is ideal for: * **Multi-tenant SaaS applications** - Soft deletes, custom fields, tenant isolation * **Event-driven architectures** - Real-time notifications via message bus * **Search-heavy applications** - Full Elasticsearch query capabilities * **High-traffic systems** - Built-in caching and performance optimizations ## Related Projects * [Foundatio](https://github.com/FoundatioFx/Foundatio) - Core building blocks (caching, messaging, queues, jobs) * [Foundatio.Parsers](https://github.com/FoundatioFx/Foundatio.Parsers) - Query parsing for dynamic filtering ## Next Steps * [Getting Started](/guide/getting-started) - Install and create your first repository * [Repository Pattern](/guide/repository-pattern) - Understand the core interfaces * [Elasticsearch Setup](/guide/elasticsearch-setup) - Configure your Elasticsearch connection --- --- url: /guide/getting-started.md --- # Getting Started This guide will help you install Foundatio.Repositories and create your first repository. ## Prerequisites * .NET 8.0 or later * Elasticsearch 7.x or 8.x (for the Elasticsearch implementation) * Basic understanding of the repository pattern ## Installation Install the NuGet packages: ::: code-group ```bash [.NET CLI] dotnet add package Foundatio.Repositories.Elasticsearch ``` ```bash [Package Manager] Install-Package Foundatio.Repositories.Elasticsearch ``` ```xml [PackageReference] ``` ::: The `Foundatio.Repositories.Elasticsearch` package includes the core `Foundatio.Repositories` package as a dependency. ## Quick Start ### 1. Define Your Entity Create a model class that implements the required interfaces: ```csharp using Foundatio.Repositories.Models; public class Employee : IIdentity, IHaveDates { public string Id { get; set; } = string.Empty; public string Name { get; set; } = string.Empty; public string Email { get; set; } = string.Empty; public int Age { get; set; } public string CompanyId { get; set; } = string.Empty; public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } } ``` **Available interfaces:** | Interface | Purpose | |-----------|---------| | `IIdentity` | Provides `Id` property (required) | | `IHaveCreatedDate` | Provides `CreatedUtc` property | | `IHaveDates` | Provides `CreatedUtc` and `UpdatedUtc` properties | | `ISupportSoftDeletes` | Provides `IsDeleted` property for soft delete support | | `IVersioned` | Provides `Version` property for optimistic concurrency | ### 2. Create an Index Configuration Define how your entity is indexed in Elasticsearch: ```csharp using Elastic.Clients.Elasticsearch.IndexManagement; using Elastic.Clients.Elasticsearch.Mapping; using Foundatio.Parsers.ElasticQueries; using Foundatio.Repositories.Elasticsearch.Configuration; using Foundatio.Repositories.Elasticsearch.Extensions; public sealed class EmployeeIndex : VersionedIndex { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 1) { } public override void ConfigureIndex(CreateIndexRequestDescriptor idx) { base.ConfigureIndex(idx.Settings(s => s .NumberOfReplicas(0) .NumberOfShards(1))); } public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.CompanyId) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) .Text(e => e.Email, t => t.AddKeywordAndSortFields()) .IntegerNumber(e => e.Age) ); } } ``` ### 3. Create the Elasticsearch Configuration Set up the connection to Elasticsearch: ```csharp using Foundatio.Repositories.Elasticsearch.Configuration; using Microsoft.Extensions.Logging; using Elastic.Transport; public class MyElasticConfiguration : ElasticConfiguration { public MyElasticConfiguration(ILoggerFactory loggerFactory) : base(loggerFactory: loggerFactory) { AddIndex(Employees = new EmployeeIndex(this)); } protected override NodePool CreateConnectionPool() { return new SingleNodePool(new Uri("http://localhost:9200")); } public EmployeeIndex Employees { get; } } ``` ### 4. Create the Repository Interface and Implementation ```csharp using Foundatio.Repositories; using Foundatio.Repositories.Elasticsearch; public interface IEmployeeRepository : ISearchableRepository { } public class EmployeeRepository : ElasticRepositoryBase, IEmployeeRepository { public EmployeeRepository(MyElasticConfiguration configuration) : base(configuration.Employees) { } } ``` ### 5. Register Services and Use Register your services with dependency injection: ```csharp using Microsoft.Extensions.DependencyInjection; var services = new ServiceCollection(); // Register logging services.AddLogging(); // Register Elasticsearch configuration services.AddSingleton(); // Register repository services.AddSingleton(); var provider = services.BuildServiceProvider(); ``` Configure the indexes (creates them if they don't exist): ```csharp var config = provider.GetRequiredService(); await config.ConfigureIndexesAsync(); ``` Use the repository: ```csharp var repository = provider.GetRequiredService(); // Add an employee var employee = await repository.AddAsync(new Employee { Name = "John Doe", Email = "john@example.com", Age = 30, CompanyId = "acme" }, o => o.ImmediateConsistency()); Console.WriteLine($"Created employee with ID: {employee.Id}"); // Find employees var results = await repository.FindAsync(q => q .FilterExpression("age:>=25") .SortExpression("name")); Console.WriteLine($"Found {results.Total} employees"); // Update an employee employee.Age = 31; await repository.SaveAsync(employee); // Delete an employee await repository.RemoveAsync(employee); ``` ## ASP.NET Core Integration For ASP.NET Core applications, configure indexes on startup: ```csharp using Foundatio.Extensions.Hosting.Startup; var builder = WebApplication.CreateBuilder(args); // Register services builder.Services.AddSingleton(); builder.Services.AddSingleton(); // Configure indexes on startup builder.Services.AddStartupAction("ConfigureIndexes", async sp => { var configuration = sp.GetRequiredService(); await configuration.ConfigureIndexesAsync(); }); var app = builder.Build(); // Wait for startup actions before serving requests app.UseWaitForStartupActionsBeforeServingRequests(); app.Run(); ``` ## Running Elasticsearch For local development, use Docker: ```bash docker run -d --name elasticsearch \ -p 9200:9200 \ -e "discovery.type=single-node" \ -e "xpack.security.enabled=false" \ docker.elastic.co/elasticsearch/elasticsearch:8.11.0 ``` Or use the provided `docker-compose.yml`: ```bash docker compose up -d ``` ## Next Steps * [Repository Pattern](/guide/repository-pattern) - Learn about the core interfaces and event handlers * [Elasticsearch Setup](/guide/elasticsearch-setup) - Advanced Elasticsearch configuration * [CRUD Operations](/guide/crud-operations) - Detailed guide to data operations * [Querying](/guide/querying) - Build dynamic queries with filters and aggregations ## ID Generation Foundatio.Repositories includes an `ObjectId` utility for generating unique, sortable IDs (similar to MongoDB's ObjectId): ```csharp using Foundatio.Repositories.Utility; // Generate a new ID string id = ObjectId.GenerateNewId().ToString(); // Example: "507f1f77bcf86cd799439011" // IDs are time-sortable var id1 = ObjectId.GenerateNewId(); var id2 = ObjectId.GenerateNewId(); // id2 > id1 (chronologically) // Extract creation time from ID var objectId = new ObjectId("507f1f77bcf86cd799439011"); DateTime createdAt = objectId.CreationTime; ``` By default, when you add a document without an ID, the repository will generate one automatically. You can customize ID generation by setting the ID before calling `AddAsync`: ```csharp var employee = new Employee { Id = ObjectId.GenerateNewId().ToString(), // Custom ID Name = "John Doe" }; await repository.AddAsync(employee); ``` --- --- url: /guide/repository-pattern.md --- # Repository Pattern Foundatio.Repositories provides a clean abstraction over data access through a hierarchy of interfaces. This guide covers the core interfaces, their methods, and the powerful event system for extending repository behavior. ## Core Interfaces ### IReadOnlyRepository\ The base interface for read-only operations: ```csharp public interface IReadOnlyRepository where T : class, new() { // Read operations Task GetByIdAsync(Id id, ICommandOptions options = null); Task> GetByIdsAsync(Ids ids, ICommandOptions options = null); Task> GetAllAsync(ICommandOptions options = null); Task ExistsAsync(Id id, ICommandOptions options = null); Task CountAsync(ICommandOptions options = null); // Cache invalidation Task InvalidateCacheAsync(T document); Task InvalidateCacheAsync(IEnumerable documents); Task InvalidateCacheAsync(string cacheKey); Task InvalidateCacheAsync(IEnumerable cacheKeys); // Events AsyncEvent> BeforeQuery { get; } AsyncEvent> AfterQuery { get; } } ``` ### IRepository\ Extends `IReadOnlyRepository` with write operations: ```csharp public interface IRepository : IReadOnlyRepository where T : class, IIdentity, new() { // Write operations Task AddAsync(T document, ICommandOptions options = null); Task AddAsync(IEnumerable documents, ICommandOptions options = null); Task SaveAsync(T document, ICommandOptions options = null); Task SaveAsync(IEnumerable documents, ICommandOptions options = null); Task PatchAsync(Id id, IPatchOperation operation, ICommandOptions options = null); Task PatchAsync(Ids ids, IPatchOperation operation, ICommandOptions options = null); Task RemoveAsync(Id id, ICommandOptions options = null); Task RemoveAsync(T document, ICommandOptions options = null); Task RemoveAsync(IEnumerable documents, ICommandOptions options = null); Task RemoveAllAsync(ICommandOptions options = null); // Events AsyncEvent> DocumentsAdding { get; } AsyncEvent> DocumentsAdded { get; } AsyncEvent> DocumentsSaving { get; } AsyncEvent> DocumentsSaved { get; } AsyncEvent> DocumentsRemoving { get; } AsyncEvent> DocumentsRemoved { get; } AsyncEvent> DocumentsChanging { get; } AsyncEvent> DocumentsChanged { get; } } ``` ### ISearchableRepository\ Extends `IRepository` with query capabilities: ```csharp public interface ISearchableRepository : IRepository, ISearchableReadOnlyRepository where T : class, IIdentity, new() { Task PatchAllAsync(IRepositoryQuery query, IPatchOperation operation, ICommandOptions options = null); Task RemoveAllAsync(IRepositoryQuery query, ICommandOptions options = null); Task BatchProcessAsync(IRepositoryQuery query, Func, Task> processFunc, ICommandOptions options = null); } ``` ### ISearchableReadOnlyRepository\ Query operations without write access: ```csharp public interface ISearchableReadOnlyRepository : IReadOnlyRepository where T : class, new() { Task> FindAsync(IRepositoryQuery query, ICommandOptions options = null); Task> FindAsAsync(IRepositoryQuery query, ICommandOptions options = null) where TResult : class, new(); Task> FindOneAsync(IRepositoryQuery query, ICommandOptions options = null); Task CountAsync(IRepositoryQuery query, ICommandOptions options = null); Task ExistsAsync(IRepositoryQuery query, ICommandOptions options = null); } ``` ## Event System The repository provides a comprehensive event system that allows you to hook into the document lifecycle. Events are fired at various stages of CRUD operations. ### Event Types | Event | When Fired | Use Cases | |-------|------------|-----------| | `DocumentsAdding` | Before documents are added | Set defaults, validate, generate IDs | | `DocumentsAdded` | After documents are added | Trigger notifications, audit logging | | `DocumentsSaving` | Before documents are saved | Track changes, validate updates | | `DocumentsSaved` | After documents are saved | Trigger notifications, audit logging | | `DocumentsRemoving` | Before documents are removed | Cascade deletes, archive data | | `DocumentsRemoved` | After documents are removed | Clean up related resources | | `DocumentsChanging` | Before any change | Universal change tracking | | `DocumentsChanged` | After any change | Universal change tracking | | `BeforeQuery` | Before query execution | Add filters, track metrics | | `AfterQuery` | After query execution | Transform aggregations, modify results | | `BeforePublishEntityChanged` | Before notification publish | Modify or cancel notifications | ### Event Argument Types #### DocumentsEventArgs\ Used for `DocumentsAdding`, `DocumentsAdded`, `DocumentsRemoving`, `DocumentsRemoved`: ```csharp public class DocumentsEventArgs : EventArgs { public IReadOnlyCollection Documents { get; } public IRepository Repository { get; } public ICommandOptions Options { get; } } ``` #### ModifiedDocumentsEventArgs\ Used for `DocumentsSaving`, `DocumentsSaved` - includes original document state: ```csharp public class ModifiedDocumentsEventArgs : EventArgs { public IReadOnlyCollection> Documents { get; } public IRepository Repository { get; } public ICommandOptions Options { get; } } public class ModifiedDocument { public T Value { get; set; } // Current/modified document public T Original { get; } // Original document before changes } ``` #### DocumentsChangeEventArgs\ Used for `DocumentsChanging`, `DocumentsChanged`: ```csharp public class DocumentsChangeEventArgs : EventArgs { public ChangeType ChangeType { get; } // Added, Saved, or Removed public IReadOnlyCollection> Documents { get; } public IRepository Repository { get; } public ICommandOptions Options { get; } } public enum ChangeType : byte { Added = 0, Saved = 1, Removed = 2 } ``` #### BeforeQueryEventArgs\ Used for `BeforeQuery`: ```csharp public class BeforeQueryEventArgs : EventArgs { public Type ResultType { get; } public IRepositoryQuery Query { get; } // Modifiable public ICommandOptions Options { get; } // Modifiable public IReadOnlyRepository Repository { get; } } ``` #### AfterQueryEventArgs\ Used for `AfterQuery` - provides access to the query result for post-processing: ```csharp public class AfterQueryEventArgs : EventArgs { public Type ResultType { get; } public IRepositoryQuery Query { get; } public ICommandOptions Options { get; } public IReadOnlyRepository Repository { get; } public CountResult Result { get; } } ``` Modify `Result.Aggregations` directly to transform aggregation results. The result is mutated in place before caching and returning. #### BeforePublishEntityChangedEventArgs\ Used for `BeforePublishEntityChanged` - supports cancellation: ```csharp public class BeforePublishEntityChangedEventArgs : CancelEventArgs { public EntityChanged Message { get; } public IReadOnlyRepository Repository { get; } // Inherited: bool Cancel { get; set; } } ``` ### Subscribing to Events #### Async Handlers Use `AddHandler` for async event handlers: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { DocumentsAdding.AddHandler(OnDocumentsAdding); DocumentsSaving.AddHandler(OnDocumentsSaving); BeforeQuery.AddHandler(OnBeforeQuery); AfterQuery.AddHandler(OnAfterQuery); } private Task OnDocumentsAdding(object sender, DocumentsEventArgs args) { foreach (var employee in args.Documents) { // Set default values if (string.IsNullOrEmpty(employee.Department)) employee.Department = "General"; } return Task.CompletedTask; } private Task OnDocumentsSaving(object sender, ModifiedDocumentsEventArgs args) { foreach (var modified in args.Documents) { var original = modified.Original; var current = modified.Value; // Detect salary changes if (original?.Salary != current.Salary) { _logger.LogInformation( "Salary changed for {Id}: {Old} -> {New}", current.Id, original?.Salary, current.Salary); } } return Task.CompletedTask; } private Task OnBeforeQuery(object sender, BeforeQueryEventArgs args) { // Add tenant filter to all queries using typed expression args.Query.FieldEquals(e => e.TenantId, _currentTenantId); return Task.CompletedTask; } private Task OnAfterQuery(object sender, AfterQueryEventArgs args) { // Flatten nested aggregation wrappers so consumers see // terms/metrics at the top level instead of inside nested buckets args.Result.Aggregations = FlattenNestedAggregations(args.Result.Aggregations); return Task.CompletedTask; } } ``` #### Sync Handlers Use `AddSyncHandler` for synchronous handlers. Returns an `IDisposable` for cleanup: ```csharp // Subscribe var subscription = repository.DocumentsAdded.AddSyncHandler((sender, args) => { Console.WriteLine($"Added {args.Documents.Count} documents"); }); // Unsubscribe when done subscription.Dispose(); ``` #### Removing Handlers For async handlers: ```csharp repository.DocumentsAdding.AddHandler(OnDocumentsAdding); // Later... repository.DocumentsAdding.RemoveHandler(OnDocumentsAdding); ``` ### Event Execution Order Events fire in a specific order during operations: ```mermaid sequenceDiagram participant App as Application participant Repo as Repository participant ES as Elasticsearch participant Bus as Message Bus App->>Repo: AddAsync(document) Repo->>Repo: DocumentsAdding Repo->>Repo: DocumentsChanging (Added) Repo->>ES: Index document ES-->>Repo: Success Repo->>Repo: DocumentsAdded Repo->>Repo: DocumentsChanged (Added) Repo->>Repo: BeforePublishEntityChanged Repo->>Bus: Publish EntityChanged Repo-->>App: Return document ``` **Add Operation:** 1. `DocumentsAdding` 2. `DocumentsChanging` (ChangeType.Added) 3. *Document indexed* 4. `DocumentsAdded` 5. `DocumentsChanged` (ChangeType.Added) 6. `BeforePublishEntityChanged` 7. *EntityChanged published* **Save Operation:** 1. `DocumentsSaving` 2. `DocumentsChanging` (ChangeType.Saved) 3. *Document indexed* 4. `DocumentsSaved` 5. `DocumentsChanged` (ChangeType.Saved) 6. `BeforePublishEntityChanged` 7. *EntityChanged published* **Remove Operation:** 1. `DocumentsRemoving` 2. `DocumentsChanging` (ChangeType.Removed) 3. *Document deleted* 4. `DocumentsRemoved` 5. `DocumentsChanged` (ChangeType.Removed) 6. `BeforePublishEntityChanged` 7. *EntityChanged published* ### Modifying Documents in Handlers Documents can be modified directly in event handlers: ```csharp private Task OnDocumentsAdding(object sender, DocumentsEventArgs args) { foreach (var employee in args.Documents) { // Set audit fields employee.CreatedBy = _currentUserId; employee.CreatedAt = DateTime.UtcNow; } return Task.CompletedTask; } private Task OnDocumentsSaving(object sender, ModifiedDocumentsEventArgs args) { foreach (var modified in args.Documents) { // Modify the Value property, not Original modified.Value.ModifiedBy = _currentUserId; modified.Value.ModifiedAt = DateTime.UtcNow; } return Task.CompletedTask; } ``` ### Canceling Operations #### Throwing Exceptions To cancel an operation, throw an exception: ```csharp private Task OnDocumentsAdding(object sender, DocumentsEventArgs args) { foreach (var employee in args.Documents) { if (string.IsNullOrEmpty(employee.Email)) throw new DocumentValidationException("Email is required"); } return Task.CompletedTask; } ``` #### Canceling Notifications Only `BeforePublishEntityChanged` supports cancellation: ```csharp BeforePublishEntityChanged.AddHandler((sender, args) => { // Don't publish notifications for internal changes if (args.Message.Data.ContainsKey("internal")) { args.Cancel = true; } return Task.CompletedTask; }); ``` ### Common Event Patterns #### Audit Logging ```csharp DocumentsChanged.AddHandler(async (sender, args) => { foreach (var doc in args.Documents) { await _auditLog.LogAsync(new AuditEntry { EntityType = typeof(Employee).Name, EntityId = doc.Value.Id, Action = args.ChangeType.ToString(), UserId = _currentUserId, Timestamp = DateTime.UtcNow, OldValue = doc.Original, NewValue = doc.Value }); } }); ``` #### Cascade Operations ```csharp DocumentsRemoving.AddHandler(async (sender, args) => { foreach (var project in args.Documents) { // Delete all tasks when project is deleted await _taskRepository.RemoveAllAsync( q => q.FieldEquals(t => t.ProjectId, project.Id)); } }); ``` #### Query Filtering ```csharp BeforeQuery.AddHandler((sender, args) => { // Always filter by tenant using typed expression args.Query.FieldEquals(e => e.TenantId, _tenantId); // Exclude archived by default args.Query.FieldEquals(e => e.IsArchived, false); return Task.CompletedTask; }); ``` ## Next Steps * [Elasticsearch Setup](/guide/elasticsearch-setup) - Configure your Elasticsearch connection * [CRUD Operations](/guide/crud-operations) - Detailed guide to data operations * [Configuration](/guide/configuration) - Repository configuration options * [Message Bus](/guide/message-bus) - Entity change notifications --- --- url: /guide/elasticsearch-setup.md --- # Elasticsearch Setup This guide covers configuring Foundatio.Repositories for Elasticsearch, including connection setup, index configuration, and advanced options. ## Elasticsearch Configuration The `ElasticConfiguration` class manages your Elasticsearch connection and indexes. ### Basic Configuration ```csharp using Elastic.Transport; using Foundatio.Repositories.Elasticsearch.Configuration; using Microsoft.Extensions.Logging; public class MyElasticConfiguration : ElasticConfiguration { public MyElasticConfiguration(ILoggerFactory loggerFactory) : base(loggerFactory: loggerFactory) { // Register indexes AddIndex(Employees = new EmployeeIndex(this)); AddIndex(Projects = new ProjectIndex(this)); } protected override NodePool CreateConnectionPool() { return new SingleNodePool(new Uri("http://localhost:9200")); } public EmployeeIndex Employees { get; } public ProjectIndex Projects { get; } } ``` ### Connection Pool Options #### Single Node For development or single-node clusters: ```csharp protected override NodePool CreateConnectionPool() { return new SingleNodePool(new Uri("http://localhost:9200")); } ``` #### Multiple Nodes For production clusters with multiple nodes: ```csharp protected override NodePool CreateConnectionPool() { var nodes = new[] { new Uri("http://es-node1:9200"), new Uri("http://es-node2:9200"), new Uri("http://es-node3:9200") }; return new StaticNodePool(nodes); } ``` #### Sniffing Connection Pool Automatically discovers cluster nodes: ```csharp protected override NodePool CreateConnectionPool() { var nodes = new[] { new Uri("http://es-node1:9200") }; return new SniffingNodePool(nodes); } ``` ### Connection Settings Override `ConfigureSettings` to customize the client: ```csharp protected override void ConfigureSettings(ElasticsearchClientSettings settings) { base.ConfigureSettings(settings); // Enable detailed logging in development if (_environment.IsDevelopment()) { settings.DisableDirectStreaming(); settings.PrettyJson(); } // Set default timeout settings.RequestTimeout(TimeSpan.FromSeconds(30)); // Configure basic authentication settings.Authentication(new BasicAuthentication("username", "password")); // Or use API key settings.Authentication(new ApiKey("encoded-api-key")); } ``` ### Serialization The new `Elastic.Clients.Elasticsearch` client uses **System.Text.Json** by default. Custom serialization is configured via `SourceSerializerFactory` if needed. ### Configuration with Dependency Injection ```csharp public class MyElasticConfiguration : ElasticConfiguration { private readonly IConfiguration _config; private readonly IWebHostEnvironment _env; public MyElasticConfiguration( IConfiguration config, IWebHostEnvironment env, ILoggerFactory loggerFactory) : base(loggerFactory: loggerFactory) { _config = config; _env = env; AddIndex(Employees = new EmployeeIndex(this)); } protected override NodePool CreateConnectionPool() { var connectionString = _config.GetConnectionString("Elasticsearch") ?? "http://localhost:9200"; return new SingleNodePool(new Uri(connectionString)); } protected override void ConfigureSettings(ElasticsearchClientSettings settings) { base.ConfigureSettings(settings); if (_env.IsDevelopment()) { settings.DisableDirectStreaming(); settings.PrettyJson(); } } public EmployeeIndex Employees { get; } } ``` ## Index Configuration ### IElasticConfiguration Interface ```csharp public interface IElasticConfiguration : IDisposable { ElasticsearchClient Client { get; } ICacheClient Cache { get; } IMessageBus MessageBus { get; } ILoggerFactory LoggerFactory { get; } IReadOnlyCollection Indexes { get; } Task ConfigureIndexesAsync(IEnumerable indexes = null); Task MaintainIndexesAsync(IEnumerable indexes = null); Task DeleteIndexesAsync(IEnumerable indexes = null); Task ReindexAsync(IEnumerable indexes = null); } ``` ### Configuring Indexes Call `ConfigureIndexesAsync` to create indexes: ```csharp var config = new MyElasticConfiguration(loggerFactory); await config.ConfigureIndexesAsync(); ``` This will: 1. Create indexes that don't exist 2. Update mappings for existing `Index` and `VersionedIndex` indexes (if compatible). Note: `DailyIndex`/`MonthlyIndex` existing partitions are **not** updated — see [Mapping Lifecycle](/guide/index-management#mapping-lifecycle). 3. Create aliases 4. Start reindexing for outdated indexes (if `beginReindexingOutdated` is true) When running at scale with multiple processes, `ConfigureIndexesAsync` uses a distributed lock and cache marker to prevent redundant Elasticsearch admin API calls. See [Index Management - Concurrency Protection](/guide/index-management#concurrency-protection) for details. ### Index Types Foundatio.Repositories provides several index types: | Type | Description | Use Case | |------|-------------|----------| | `Index` | Basic index | Simple entities | | `VersionedIndex` | Schema versioning | Evolving schemas | | `DailyIndex` | Daily partitioning | Time-series data | | `MonthlyIndex` | Monthly partitioning | Time-series data | See [Index Management](/guide/index-management) for detailed documentation. ## Index Mapping ### Basic Mapping ```csharp public sealed class EmployeeIndex : VersionedIndex { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 1) { } public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) // Disable dynamic mapping .Properties(p => p .SetupDefaults() // Configure Id, CreatedUtc, UpdatedUtc, IsDeleted .Keyword(e => e.CompanyId) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) .IntegerNumber(e => e.Age) ); } } ``` ::: warning Dynamic mapping is disabled All index configurations should use `.Dynamic(false)`. This means any model field you want to query, filter, sort, or aggregate on **must** have an explicit mapping in `ConfigureIndexMapping`. Unmapped fields are stored in `_source` but never indexed -- queries against them silently return zero results with no error. For details on how and when mappings are applied (including important differences between `VersionedIndex` and `DailyIndex`/`MonthlyIndex`), see [Mapping Lifecycle](/guide/index-management#mapping-lifecycle). ::: ### SetupDefaults Extension The `SetupDefaults()` extension automatically configures common fields: ```csharp .Properties(p => p.SetupDefaults()) ``` This configures: * `Id` as keyword * `CreatedUtc` as date * `UpdatedUtc` as date * `IsDeleted` as boolean (if `ISupportSoftDeletes`) * `Version` as keyword (if `IVersioned`) ### Field Types #### Keyword Fields For exact matching and aggregations: ```csharp .Keyword(e => e.Status) ``` #### Text Fields with Keywords For full-text search with exact matching: ```csharp .Text(e => e.Name, t => t.AddKeywordAndSortFields()) ``` This creates: * `name` - Analyzed text field * `name.keyword` - Exact match keyword field * `name.sort` - Normalized for sorting #### Nested Objects ```csharp .Nested(e => e.Addresses, n => n .Properties(ap => ap .Keyword(a => a.City) .Keyword(a => a.Country) )) ``` Fields mapped as `nested` are automatically wrapped in Elasticsearch `nested` queries and `nested` aggregations when queried through filter or aggregation expressions. See [Nested Queries](/guide/querying#nested-queries) and [Nested Field Aggregations](/guide/querying#nested-field-aggregations) for details and examples. ### Index Settings ```csharp public override void ConfigureIndex(CreateIndexRequestDescriptor idx) { base.ConfigureIndex(idx.Settings(s => s .NumberOfShards(3) .NumberOfReplicas(1) .Analysis(a => a .AddSortNormalizer() ))); } ``` ## Caching and Messaging ### Adding Cache Support ```csharp using Foundatio.Caching; public class MyElasticConfiguration : ElasticConfiguration { public MyElasticConfiguration( ICacheClient cache, ILoggerFactory loggerFactory) : base(cache: cache, loggerFactory: loggerFactory) { AddIndex(Employees = new EmployeeIndex(this)); } // ... } ``` ### Adding Message Bus Support ```csharp using Foundatio.Messaging; public class MyElasticConfiguration : ElasticConfiguration { public MyElasticConfiguration( ICacheClient cache, IMessageBus messageBus, ILoggerFactory loggerFactory) : base(cache: cache, messageBus: messageBus, loggerFactory: loggerFactory) { AddIndex(Employees = new EmployeeIndex(this)); } // ... } ``` ### Full Configuration Example ```csharp public class MyElasticConfiguration : ElasticConfiguration { private readonly IConfiguration _config; private readonly IWebHostEnvironment _env; public MyElasticConfiguration( IConfiguration config, IWebHostEnvironment env, ICacheClient cache, IMessageBus messageBus, ILoggerFactory loggerFactory) : base(cache: cache, messageBus: messageBus, loggerFactory: loggerFactory) { _config = config; _env = env; AddIndex(Employees = new EmployeeIndex(this)); AddIndex(Projects = new ProjectIndex(this)); AddIndex(AuditLogs = new AuditLogIndex(this)); } protected override NodePool CreateConnectionPool() { var connectionString = _config.GetConnectionString("Elasticsearch"); if (string.IsNullOrEmpty(connectionString)) connectionString = "http://localhost:9200"; var uris = connectionString.Split(',').Select(s => new Uri(s.Trim())); if (uris.Count() == 1) return new SingleNodePool(uris.First()); return new StaticNodePool(uris); } protected override void ConfigureSettings(ElasticsearchClientSettings settings) { base.ConfigureSettings(settings); if (_env.IsDevelopment()) { settings.DisableDirectStreaming(); settings.PrettyJson(); } var username = _config["Elasticsearch:Username"]; var password = _config["Elasticsearch:Password"]; if (!string.IsNullOrEmpty(username)) settings.Authentication(new BasicAuthentication(username, password)); } public EmployeeIndex Employees { get; } public ProjectIndex Projects { get; } public AuditLogIndex AuditLogs { get; } } ``` ## Dependency Injection Registration ### Basic Registration ```csharp services.AddSingleton(); services.AddSingleton(); ``` ### With Foundatio Services ```csharp // Register Foundatio services services.AddSingleton(new InMemoryCacheClient()); services.AddSingleton(new InMemoryMessageBus()); // Register Elasticsearch configuration services.AddSingleton(); // Register repositories services.AddSingleton(); services.AddSingleton(); ``` ### Startup Configuration ```csharp // In Program.cs or Startup.cs var config = app.Services.GetRequiredService(); await config.ConfigureIndexesAsync(); ``` Or use a startup action: ```csharp services.AddStartupAction("ConfigureElasticsearch", async sp => { var config = sp.GetRequiredService(); await config.ConfigureIndexesAsync(); }); ``` ## Parent-Child Relationships Elasticsearch supports parent-child relationships using join fields. This allows you to model hierarchical data where children are stored in the same index as parents but can be queried independently. ### Defining Parent-Child Documents Implement `IParentChildDocument` for both parent and child entities: ```csharp using Foundatio.Repositories.Elasticsearch; using Elastic.Clients.Elasticsearch; using Foundatio.Repositories.Elasticsearch.Repositories; using Foundatio.Repositories.Models; // Parent document public class Organization : IParentChildDocument, IHaveDates, ISupportSoftDeletes { public string Id { get; set; } // IParentChildDocument - parent doesn't need a ParentId string IParentChildDocument.ParentId { get; set; } JoinField IParentChildDocument.Discriminator { get; set; } public string Name { get; set; } public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } public bool IsDeleted { get; set; } } // Child document public class Employee : IParentChildDocument, IHaveDates, ISupportSoftDeletes { public string Id { get; set; } // Child must have ParentId public string ParentId { get; set; } JoinField IParentChildDocument.Discriminator { get; set; } public string Name { get; set; } public string Email { get; set; } public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } public bool IsDeleted { get; set; } } ``` ### Configuring the Index Create a single index with a join field mapping: ```csharp public sealed class OrganizationIndex : VersionedIndex { public OrganizationIndex(IElasticConfiguration configuration) : base(configuration, "organizations", version: 1) { } public override void ConfigureIndex(CreateIndexRequestDescriptor idx) { base.ConfigureIndex(idx .Settings(s => s.NumberOfReplicas(0).NumberOfShards(1)) .Mappings(m => m .Properties(p => p .SetupDefaults() .Keyword(o => ((Organization)o).Name) .Keyword(e => ((Employee)e).Email) .Join(d => d.Discriminator, j => j .Relations(r => r.Add("organization", new[] { "employee" })) ) ))); } } ``` ### Creating Repositories Create separate repositories for parent and child: ```csharp // Parent repository public class OrganizationRepository : ElasticRepositoryBase { public OrganizationRepository(OrganizationIndex index) : base(index) { } } // Child repository - must set HasParent and GetParentIdFunc public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(OrganizationIndex index) : base(index) { HasParent = true; GetParentIdFunc = e => e.ParentId; // Required for soft delete filtering on parent DocumentType = typeof(Employee); ParentDocumentType = typeof(Organization); } } ``` ### Working with Parent-Child Documents ```csharp // Add parent var org = await orgRepository.AddAsync(new Organization { Name = "Acme Corp" }); // Add child with parent reference var employee = await employeeRepository.AddAsync(new Employee { Name = "John Doe", Email = "john@acme.com", ParentId = org.Id // Link to parent }); // Get child by ID (requires routing for efficiency) var emp = await employeeRepository.GetByIdAsync(new Id(employee.Id, org.Id)); // Or without routing (uses search fallback) var emp = await employeeRepository.GetByIdAsync(employee.Id); // Query children by parent var employees = await employeeRepository.FindAsync(q => q.ParentId("organization", org.Id)); ``` ### Parent-Child Soft Delete Behavior When a parent is soft-deleted, children are automatically filtered from queries: ```csharp // Soft delete the parent org.IsDeleted = true; await orgRepository.SaveAsync(org); // Children are now filtered (even though they're not deleted) var count = await employeeRepository.CountAsync(); // Returns 0 // Restore parent org.IsDeleted = false; await orgRepository.SaveAsync(org); // Children are visible again var count = await employeeRepository.CountAsync(); // Returns children count ``` ### Querying with Parent Filters ```csharp // Find children where parent matches criteria var results = await employeeRepository.FindAsync(q => q .ParentQuery(pq => pq .DocumentType() .FieldEquals(o => o.Name, "Acme Corp"))); ``` ::: warning Routing Considerations * Child documents are routed to the same shard as their parent using `ParentId` * For best performance, always provide routing when getting children by ID: `new Id(childId, parentId)` * Without routing, the repository falls back to a search query which is slower ::: ## Health Checks Add Elasticsearch health checks: ```csharp services.AddHealthChecks() .AddCheck("elasticsearch", () => { var config = services.BuildServiceProvider() .GetRequiredService(); var response = config.Client.Ping(); return response.IsValidResponse ? HealthCheckResult.Healthy() : HealthCheckResult.Unhealthy("Elasticsearch is not responding"); }); ``` ## Next Steps * [CRUD Operations](/guide/crud-operations) - Working with documents * [Querying](/guide/querying) - Building queries * [Index Management](/guide/index-management) - Advanced index configuration * [Configuration](/guide/configuration) - Repository configuration options --- --- url: /guide/crud-operations.md --- # CRUD Operations This guide covers the core Create, Read, Update, and Delete operations in Foundatio.Repositories. ## Adding Documents ### Add Single Document ```csharp var employee = new Employee { Name = "John Doe", Email = "john@example.com", Age = 30 }; var result = await repository.AddAsync(employee); Console.WriteLine($"Created with ID: {result.Id}"); ``` The repository automatically: * Generates an ID if not provided * Sets `CreatedUtc` and `UpdatedUtc` (if `IHaveDates`) * Validates the document (if validation is configured) * Publishes `EntityChanged` notification (if message bus is configured) ### Add Multiple Documents ```csharp var employees = new List { new Employee { Name = "John Doe", Age = 30 }, new Employee { Name = "Jane Smith", Age = 28 } }; await repository.AddAsync(employees); ``` ### Add with Options ```csharp // Immediate consistency - wait for index refresh var employee = await repository.AddAsync(entity, o => o.ImmediateConsistency()); // Disable notifications await repository.AddAsync(entity, o => o.Notifications(false)); // Enable caching await repository.AddAsync(entity, o => o.Cache()); // Combine options await repository.AddAsync(entity, o => o .ImmediateConsistency() .Cache() .Notifications(false)); ``` ## Reading Documents ### Get by ID ```csharp var employee = await repository.GetByIdAsync("employee-123"); if (employee == null) { Console.WriteLine("Employee not found"); } ``` ### Get Multiple by IDs ```csharp var ids = new[] { "emp-1", "emp-2", "emp-3" }; var employees = await repository.GetByIdsAsync(ids); Console.WriteLine($"Found {employees.Count} employees"); ``` ### Get All Documents ```csharp var results = await repository.GetAllAsync(); Console.WriteLine($"Total: {results.Total}"); foreach (var employee in results.Documents) { Console.WriteLine($"- {employee.Name}"); } ``` ### Check Existence ```csharp bool exists = await repository.ExistsAsync("employee-123"); ``` ### Count Documents ```csharp long count = await repository.CountAsync(); ``` ### Read with Options ```csharp // With caching var employee = await repository.GetByIdAsync(id, o => o.Cache()); // Include soft-deleted documents var employee = await repository.GetByIdAsync(id, o => o.IncludeSoftDeletes()); // Select specific fields var employee = await repository.GetByIdAsync(id, o => o .Include(e => e.Name) .Include(e => e.Email)); ``` ## Updating Documents ### Save (Full Update) ```csharp var employee = await repository.GetByIdAsync(id); employee.Name = "John Smith"; employee.Age = 31; await repository.SaveAsync(employee); ``` The repository automatically: * Updates `UpdatedUtc` (if `IHaveDates`) * Checks version for conflicts (if `IVersioned`) * Invalidates cache * Publishes `EntityChanged` notification ::: tip Consistent Date Tracking All write operations — `AddAsync`, `SaveAsync`, and all patch types (`PatchAsync`, `PatchAllAsync`) — automatically set `UpdatedUtc` for models implementing `IHaveDates`. `CreatedUtc` is set on initial creation and is not changed by later operations unless the existing value is missing or invalid (for example, `DateTime.MinValue` or a timestamp in the future). For `ScriptPatch` and `PartialPatch`, if you explicitly provide the `updatedUtc` field, the framework respects your value. `JsonPatch` and `ActionPatch` always overwrite `UpdatedUtc`, matching `SaveAsync` semantics. ::: ### Save Multiple Documents ```csharp var employees = await repository.GetByIdsAsync(ids); foreach (var emp in employees) { emp.Department = "Engineering"; } await repository.SaveAsync(employees); ``` ### Save with Options ```csharp // Skip version check await repository.SaveAsync(employee, o => o.SkipVersionCheck()); // Immediate consistency await repository.SaveAsync(employee, o => o.ImmediateConsistency()); // Provide original for change detection await repository.SaveAsync(employee, o => o.AddOriginals(originalEmployee)); ``` ## Deleting Documents ### Remove by ID ```csharp await repository.RemoveAsync("employee-123"); ``` ### Remove Document ```csharp var employee = await repository.GetByIdAsync(id); await repository.RemoveAsync(employee); ``` ### Remove Multiple Documents ```csharp var employees = await repository.GetByIdsAsync(ids); await repository.RemoveAsync(employees); ``` ### Remove All Documents ::: warning This permanently deletes ALL documents in the index. ::: ```csharp long deleted = await repository.RemoveAllAsync(); Console.WriteLine($"Deleted {deleted} documents"); ``` ### Remove with Query For `ISearchableRepository`: ```csharp // Remove all employees in a department long deleted = await repository.RemoveAllAsync( q => q.FieldEquals(e => e.Department, "Sales")); ``` ::: tip Version Conflicts and Retries When no event listeners are registered and caching is disabled, `RemoveAllAsync` uses Elasticsearch's `delete_by_query`. This API snapshots document versions when it begins and skips any document modified before the delete executes, counting it as a version conflict. Because `delete_by_query` does not support `retry_on_conflict`, the repository automatically re-runs the query up to the configured retry count (default `10`, override with `o.Retry(n)`) until no conflicts remain. The returned count is the cumulative number of documents deleted across all attempts. If conflicts persist after the retry budget is exhausted, a warning is logged and the partial count is returned; no exception is thrown. ::: ### Soft Delete vs Hard Delete If your entity implements `ISupportSoftDeletes`: ```csharp // Soft delete - sets IsDeleted = true employee.IsDeleted = true; await repository.SaveAsync(employee); // Hard delete - permanently removes await repository.RemoveAsync(employee); ``` See [Soft Deletes](/guide/soft-deletes) for more details. ## Patch Operations Patch operations allow partial updates without fetching the full document. ### Partial Patch Update specific fields: ```csharp await repository.PatchAsync(id, new PartialPatch(new { Age = 32 })); ``` ### Script Patch Use Elasticsearch Painless scripts: ```csharp await repository.PatchAsync(id, new ScriptPatch("ctx._source.counter += params.increment") { Params = new Dictionary { ["increment"] = 1 } }); ``` ### JSON Patch RFC 6902 JSON Patch operations: ```csharp var patch = new PatchDocument( new ReplaceOperation { Path = "name", Value = "John Smith" }, new AddOperation { Path = "tags/-", Value = "senior" } ); await repository.PatchAsync(id, new JsonPatch(patch)); ``` ### Action Patch Lambda-based patching: ```csharp await repository.PatchAsync(id, new ActionPatch(e => { e.Name = "John Smith"; e.Age = 32; })); ``` ### Bulk Patch Patch multiple documents by query: ```csharp // Increment counter for all employees in department await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Engineering"), new ScriptPatch("ctx._source.reviewCount++")); ``` See [Patch Operations](/guide/patch-operations) for more details. ## Batch Processing Process large datasets in batches: ```csharp long processed = await repository.BatchProcessAsync( q => q.FieldEquals(e => e.Status, "pending"), async batch => { foreach (var employee in batch.Documents) { // Process each employee await ProcessEmployeeAsync(employee); } return true; // Continue processing }, o => o.PageLimit(100)); Console.WriteLine($"Processed {processed} employees"); ``` Return `false` from the callback to stop processing early: ```csharp int count = 0; await repository.BatchProcessAsync(query, async batch => { count += batch.Documents.Count; return count < 1000; // Stop after 1000 documents }); ``` ## Find Results Query operations return `FindResults`: ```csharp public class FindResults { public IReadOnlyCollection Documents { get; } public IReadOnlyCollection> Hits { get; } public long Total { get; } public int Page { get; } public bool HasMore { get; } // Automatic pagination public Task NextPageAsync(); } ``` ### Iterating Results ```csharp var results = await repository.FindAsync(query); // Access documents directly foreach (var employee in results.Documents) { Console.WriteLine(employee.Name); } // Access hits for scores and metadata foreach (var hit in results.Hits) { Console.WriteLine($"{hit.Document.Name} (score: {hit.Score})"); } ``` ### Automatic Pagination ```csharp var results = await repository.FindAsync(query, o => o.PageLimit(100)); do { foreach (var employee in results.Documents) { await ProcessAsync(employee); } } while (await results.NextPageAsync()); ``` ## Command Options Reference | Option | Description | |--------|-------------| | `.ImmediateConsistency()` | Wait for index refresh | | `.Consistency(mode)` | Set consistency mode | | `.Cache()` | Enable caching | | `.CacheKey(key)` | Set cache key | | `.CacheExpiresIn(duration)` | Set cache expiration | | `.Notifications(bool)` | Enable/disable notifications | | `.SkipValidation()` | Skip document validation | | `.SkipVersionCheck()` | Skip optimistic concurrency | | `.PageLimit(limit)` | Set page size | | `.PageNumber(page)` | Set page number | | `.Include(field)` | Include specific field | | `.Exclude(field)` | Exclude specific field | | `.IncludeSoftDeletes()` | Include soft-deleted documents | | `.SoftDeleteMode(mode)` | Set soft delete query mode | ## Error Handling ### Common Exceptions ```csharp try { await repository.SaveAsync(employee); } catch (DocumentNotFoundException ex) { // Document doesn't exist Console.WriteLine($"Document {ex.Id} not found"); } catch (VersionConflictDocumentException ex) { // Optimistic concurrency conflict Console.WriteLine($"Version conflict: {ex.Message}"); } catch (DocumentValidationException ex) { // Validation failed Console.WriteLine($"Validation error: {ex.Message}"); } catch (DuplicateDocumentException ex) { // Duplicate document Console.WriteLine($"Duplicate document: {ex.Message}"); } ``` ### Retry Pattern ```csharp var employee = await repository.GetByIdAsync(id); int retries = 3; while (retries > 0) { try { employee.Counter++; await repository.SaveAsync(employee); break; } catch (VersionConflictDocumentException) { retries--; if (retries == 0) throw; // Refresh and retry employee = await repository.GetByIdAsync(id); } } ``` ## Partial Failure Behavior When adding or saving multiple documents in a single call, some documents may succeed while others fail. The repository processes successes before throwing an exception for failures. ### How It Works 1. **Successful documents are fully processed** — events are fired, cache is populated, and change notifications are sent. 2. **Failed documents leave cache unchanged** — failed writes don't mutate Elasticsearch, so existing cache entries remain valid. Cache consistency for concurrent writes is handled by message bus notifications. 3. **A typed exception is thrown** after all successes are processed. ### Exception Types by Operation | Operation | Failure Cause | Exception | |-----------|---------------|-----------| | `AddAsync` | Document ID already exists | `DuplicateDocumentException` | | `SaveAsync` | Version conflict | `VersionConflictDocumentException` | | `PatchAsync` | Version conflict | `VersionConflictDocumentException` | | `PatchAsync` | Document not found | `DocumentNotFoundException` | | Any | Other Elasticsearch error | `DocumentException` | ### Example ```csharp try { await repository.AddAsync(employees); } catch (DuplicateDocumentException ex) { // Some documents were added successfully (events fired, cached, notified). // Duplicate documents preserve their existing cache entries (nothing was mutated). // ex.Message contains details about which documents failed. _logger.LogWarning(ex, "Some documents already existed"); } ``` ### Automatic Retry Behavior The repository includes a resilience policy for transient Elasticsearch errors: * **HTTP 429 (Too Many Requests)** and **503 (Service Unavailable)** are automatically retried with exponential backoff (up to 3 retries). * **Version conflicts (409)** on `AddAsync`/`SaveAsync` are **not** retried — the caller should handle these. * `DuplicateDocumentException` is **not** retried by the resilience policy. **Bulk delete retry**: Multi-document `RemoveAsync` (bulk delete) has its own dedicated retry loop — up to 3 retries with exponential backoff (1s, 2s, 4s) for transient 429/503 errors. Successfully deleted documents from earlier attempts are processed (events, cache, notifications) even if later retries fail for remaining documents. Fatal and conflict (HTTP 409) item IDs are accumulated across all attempts; remaining retryable errors and transport failures are surfaced through the final `BulkResult`, and any final error state (`HasErrors`) is thrown after the loop. ::: tip For operations where version conflicts are expected (e.g., high-contention counters), use `ScriptPatch` with `RetryOnConflict` instead of `SaveAsync`. Script patches are executed atomically on the Elasticsearch node and can be retried server-side. ::: ## Next Steps * [Querying](/guide/querying) - Build dynamic queries * [Patch Operations](/guide/patch-operations) - Advanced patching * [Caching](/guide/caching) - Cache configuration * [Soft Deletes](/guide/soft-deletes) - Soft delete behavior --- --- url: /guide/querying.md --- # Querying Foundatio.Repositories provides powerful querying capabilities through `ISearchableRepository`. The query system is built on [Foundatio.Parsers](https://github.com/FoundatioFx/Foundatio.Parsers), which provides Lucene-style query parsing with support for filtering, sorting, and aggregations. ## Query Parser (Foundatio.Parsers) The query expressions used throughout this library are powered by Foundatio.Parsers, which translates human-readable query strings into Elasticsearch queries. Key features include: * **Lucene-style syntax** - Familiar query syntax for developers * **Field aliasing** - Map user-friendly names to actual field names * **Type coercion** - Automatic type conversion for dates, numbers, etc. * **Validation** - Query validation against index mappings * **Extensibility** - Custom query visitors and field resolvers ## Basic Queries ### Find with Filter Expression Use Lucene-style filter expressions: ```csharp // Simple field match var results = await repository.FindAsync(q => q.FilterExpression("age:30")); // Range queries var results = await repository.FindAsync(q => q.FilterExpression("age:>=25")); var results = await repository.FindAsync(q => q.FilterExpression("age:[25 TO 35]")); // Multiple conditions (AND) var results = await repository.FindAsync(q => q.FilterExpression("age:>=25 AND department:Engineering")); // OR conditions var results = await repository.FindAsync(q => q.FilterExpression("department:Engineering OR department:Sales")); // NOT conditions var results = await repository.FindAsync(q => q.FilterExpression("NOT status:inactive")); // Wildcards var results = await repository.FindAsync(q => q.FilterExpression("name:John*")); // Exists check var results = await repository.FindAsync(q => q.FilterExpression("_exists_:email")); ``` ::: tip Prefer Strongly-Typed Queries Filter expressions are convenient for dynamic or user-supplied queries, but for application code we recommend using [Strongly-Typed Queries](#strongly-typed-queries) whenever possible. Strongly-typed queries provide compile-time field name checking, full IDE support (Find References, Rename/Refactor, Go to Definition), and runtime field-type validation that catches misuse like `FieldEquals` on text-only fields with a `QueryValidationException`. ::: ### Find with Search Expression Full-text search across analyzed fields: ```csharp var results = await repository.FindAsync(q => q.SearchExpression("john developer")); ``` #### Default Search Fields `SearchExpression` generates a multi-match query across a set of **default search fields**. Without explicit configuration, Elasticsearch uses its own `index.query.default_field` setting (typically `*`, which matches all top-level fields). You can control exactly which fields are searched by overriding `ConfigureQueryParser` on your `Index` class: ```csharp public sealed class EmployeeIndex : Index { // ... constructor, ConfigureIndex, ConfigureIndexMapping ... protected override void ConfigureQueryParser(ElasticQueryParserConfiguration config) { base.ConfigureQueryParser(config); config.SetDefaultFields([ nameof(Employee.Name).ToLowerInvariant(), nameof(Employee.EmailAddress).ToLowerInvariant() ]); } } ``` With this configuration, `SearchExpression("john")` generates a multi-match query targeting only `name` and `emailAddress` rather than every field in the index. #### Alternative: CopyTo with a Catch-All Field Instead of multi-match, you can copy field values into a single analyzed field at index time using `CopyTo`: ```csharp public override TypeMappingDescriptor ConfigureIndexMapping( TypeMappingDescriptor map) { return map .Dynamic(false) .Properties(p => p .SetupDefaults() .Text(f => f.Name("_all")) .Text(f => f.Name(e => e.Name).AddKeywordAndSortFields() .CopyTo(c => c.Field("_all"))) .Keyword(f => f.Name(e => e.EmailAddress) .CopyTo(c => c.Field("_all"))) ); } ``` | Approach | Trade-offs | |---|---| | `SetDefaultFields` | No extra index storage; generates a multi-match query at search time. Keeps the index lean. | | `CopyTo` catch-all | Increases index size (field values stored twice); single-field match at search time, which can be faster for high-throughput queries. | Both approaches can be combined. For nested field support in default search fields, see [Default Fields with Nested Paths](#default-fields-with-nested-paths) in the Nested Queries section. ### Find One Get a single matching document: ```csharp var hit = await repository.FindOneAsync(q => q.FieldEquals(e => e.Email, "john@example.com")); var employee = hit?.Document; ``` ## Strongly-Typed Queries ### Field Equals ```csharp // Single value var results = await repository.FindAsync(q => q.FieldEquals(e => e.Department, "Engineering")); // Multiple values (OR) var results = await repository.FindAsync(q => q.FieldEquals(e => e.Status, "active", "pending")); // Enum values var results = await repository.FindAsync(q => q.FieldEquals(e => e.Type, EmployeeType.FullTime)); ``` ### Field Conditions `FieldCondition` supports equality, text matching, existence checks, and range comparisons: ```csharp // Equality check var results = await repository.FindAsync(q => q .FieldCondition(e => e.Name, ComparisonOperator.Equals, "John Smith")); // Contains (for text fields) var results = await repository.FindAsync(q => q .FieldCondition(e => e.Name, ComparisonOperator.Contains, "John")); ``` Available operators: `Equals`, `NotEquals`, `IsEmpty`, `HasValue`, `Contains`, `NotContains`, `GreaterThan`, `GreaterThanOrEqual`, `LessThan`, `LessThanOrEqual`. ### Range Operators For one-sided comparisons and non-date types, use the range operator shorthands: ```csharp // Date comparison (generates DateRangeQuery) var results = await repository.FindAsync(q => q .FieldLessThanOrEqual(e => e.SnoozeUntilUtc, DateTime.UtcNow)); // Numeric comparison (generates NumericRangeQuery) var results = await repository.FindAsync(q => q .FieldGreaterThanOrEqual(e => e.Age, 18)); // Bounded range (two conditions ANDed) var results = await repository.FindAsync(q => q .FieldGreaterThanOrEqual(e => e.Age, 18) .FieldLessThan(e => e.Age, 65)); // String/keyword comparison (generates TermRangeQuery on keyword field) var results = await repository.FindAsync(q => q .FieldGreaterThanOrEqual(e => e.InstanceKey, "20240101")); // Conditional range (only applied when condition is true) var results = await repository.FindAsync(q => q .FieldGreaterThanIf(e => e.Age, minAge, minAge is not null)); ``` > **String ranges require keyword fields:** String range operators generate a `TermRangeQuery` and automatically resolve to the `.keyword` sub-field (like `FieldEquals`). If the field is an analyzed text field with no `.keyword` sub-field, a `QueryValidationException` is thrown at build time. **DateRange vs range operators:** Use `.DateRange(start, end, field)` for bounded date windows (validates start < end, supports timezone). Use `FieldGreaterThan`/`FieldLessThanOrEqual` etc. for one-sided comparisons and non-date types. > **Numeric precision note:** `long` values use NEST's `LongRangeQuery` which preserves full precision. `decimal` values are converted to `double` for NEST's `NumericRangeQuery`, which may lose precision for values exceeding ~15-17 significant digits. If exact precision matters, prefer `long` fields with an explicit scaling factor. ### Contains (Full-Text Token Matching) `FieldContains` generates a `MatchQuery` on analyzed fields. It matches complete analyzed tokens, NOT substrings or wildcards: ```csharp // Matches documents where Name contains the token "eric" var results = await repository.FindAsync(q => q .FieldContains(e => e.Name, "Eric")); // Multiple tokens: all must be present (AND), order-independent // Matches "Eric J. Smith" AND "Smith, Eric" but NOT "Eric" var results = await repository.FindAsync(q => q .FieldContains(e => e.Name, "Eric Smith")); // DOES NOT WORK: "Er" is not a complete token var results = await repository.FindAsync(q => q .FieldContains(e => e.Name, "Er")); // returns nothing ``` ### OR / AND / NOT Grouping For complex boolean logic, use `FieldOr`, `FieldAnd`, and `FieldNot`: ```csharp // Simple OR: match either condition var results = await repository.FindAsync(q => q.FieldOr(g => g .FieldEquals(f => f.IsPrivate, false) .FieldEquals(f => f.CompanyId, companyIds) )); // Nested AND inside OR var results = await repository.FindAsync(q => q.FieldOr(g => g .FieldEquals(f => f.Status, "RunNow") .FieldAnd(g2 => g2 .FieldEquals(f => f.IsEnabled, true) .FieldLessThanOrEqual(f => f.NextRunDateUtc, DateTime.UtcNow) ) )); // NOT: exclude documents matching any condition (AND-NOT semantics) var results = await repository.FindAsync(q => q.FieldNot(g => g .FieldEquals(f => f.BillingStatus, BillingStatus.Active) .FieldEquals(f => f.BillingStatus, BillingStatus.Trialing) )); // Dynamic/conditional OR groups (builder API) var group = FieldConditionGroup.Or(); group.FieldEquals(f => f.IsPrivate, false); if (privateIds.Count > 0) group.FieldEquals(f => f.PrivateId, privateIds); if (includeIds.Count > 0) group.FieldEquals(f => f.Id, includeIds); var results = await repository.FindAsync(q => q.FieldOr(group)); ``` > **FieldNot semantics:** Multiple conditions inside `FieldNot` produce NOT A AND NOT B (exclude documents matching **any** clause). For NOT (A AND B), nest an explicit AND: `FieldNot(g => g.FieldAnd(g2 => g2.FieldEquals(A).FieldEquals(B)))`. ### Field-Type Validation The query builder performs runtime validation and throws `QueryValidationException` for detectable misuse: * **FieldEquals/FieldNotEquals on text-only fields** — throws if the field is an analyzed text field with no `.keyword` sub-field (TermQuery on analyzed fields almost never matches). * **FieldContains/FieldNotContains on keyword fields** — throws because MatchQuery requires an analyzed field. * **FieldEquals on IsDeleted with ActiveOnly soft-delete mode** — throws because it creates a contradictory filter. * **Range with null value** — throws with guidance to use `*If` variants or `FieldHasValue`/`FieldEmpty`. * **Range with collection value** — throws with guidance to use `FieldEquals` for multi-value matching. > **Note:** For wildcard/prefix matching on keyword fields, use `FilterExpression("field:pattern*")`. For phrase matching (adjacent words in order), use `FilterExpression("field:\"quick brown\"")`. ### Field Empty/Has Value ```csharp // Find documents where field is null or empty var results = await repository.FindAsync(q => q.FieldEmpty(e => e.ManagerId)); // Find documents where field has a value var results = await repository.FindAsync(q => q.FieldHasValue(e => e.ManagerId)); ``` ### Date Range `.DateRange()` adds an Elasticsearch filter clause that restricts documents by a date field value. It does **not** select which physical indexes are queried — it only filters documents within whatever indexes are targeted. ```csharp var results = await repository.FindAsync(q => q .DateRange( start: DateTime.UtcNow.AddDays(-30), end: DateTime.UtcNow, field: e => e.CreatedUtc)); ``` #### Time-Series Indexes (DailyIndex / MonthlyIndex) When querying a `DailyIndex` or `MonthlyIndex`, each partition (day or month) is a separate physical Elasticsearch index. Without specifying which indexes to target, the query runs against the umbrella alias and scans all partitions regardless of the date range filter. To limit which physical indexes are queried, use `.Index(start, end)` alongside `.DateRange()`: ```csharp var start = DateTime.UtcNow.AddDays(-7); var end = DateTime.UtcNow; var results = await repository.FindAsync(q => q .Index(start, end) // target only the relevant daily index partitions .DateRange(start, end, e => e.CreatedUtc) // filter documents within those indexes ); ``` > **Note:** `.Index(start, end)` and `.DateRange()` serve different purposes and must be set independently. `DateRange` without `.Index()` is still valid — it will filter documents correctly — but it queries all partitions, which is less efficient for large time-series datasets. #### Large Range Fallback To prevent generating an excessively long list of individual index names, index selection falls back to the alias (which covers all partitions) when the requested range is too broad: | Index type | Threshold | Behavior | |---|---|---| | `DailyIndex` | Range >= 3 months, or exceeds `MaxIndexAge` | Falls back to alias | | `MonthlyIndex` | Range > 1 year, or exceeds `MaxIndexAge` | Falls back to alias | The query is still executed correctly in the fallback case — Elasticsearch simply searches all partitions — but there is no index pruning optimization. The `.DateRange()` filter still narrows the returned documents. ### ID Queries ```csharp // Find by IDs var results = await repository.FindAsync(q => q.Id("emp-1", "emp-2", "emp-3")); // Exclude IDs var results = await repository.FindAsync(q => q.ExcludedId("emp-1")); ``` ## Sorting ### Sort Expression ```csharp // Single field ascending var results = await repository.FindAsync(q => q.SortExpression("name")); // Descending var results = await repository.FindAsync(q => q.SortExpression("-createdUtc")); // Multiple fields var results = await repository.FindAsync(q => q.SortExpression("department -salary")); ``` ### Strongly-Typed Sort ```csharp var results = await repository.FindAsync(q => q .SortAscending(e => e.Name) .SortDescending(e => e.CreatedUtc)); ``` ## Pagination ### Basic Pagination ```csharp var results = await repository.FindAsync( q => q.FieldEquals(e => e.Department, "Engineering"), o => o.PageNumber(1).PageLimit(25)); Console.WriteLine($"Page {results.Page}, Total: {results.Total}, HasMore: {results.HasMore}"); ``` ### Automatic Pagination ```csharp var results = await repository.FindAsync(query, o => o.PageLimit(100)); do { foreach (var doc in results.Documents) { await ProcessAsync(doc); } } while (await results.NextPageAsync()); ``` ### Snapshot Paging (Scroll API) For large result sets, use snapshot paging: ```csharp var results = await repository.FindAsync( query, o => o.SnapshotPaging().SnapshotPagingLifetime(TimeSpan.FromMinutes(5))); do { foreach (var doc in results.Documents) { await ProcessAsync(doc); } } while (await results.NextPageAsync()); ``` ### Search After Paging More efficient for deep pagination: ```csharp var results = await repository.FindAsync( q => q.SortDescending(e => e.CreatedUtc), o => o.SearchAfterPaging()); // For subsequent pages, use the token var nextResults = await repository.FindAsync( q => q.SortDescending(e => e.CreatedUtc), o => o.SearchAfterToken(results.GetSearchAfterToken())); ``` > \[!WARNING] > **Avoid unstable sort keys (`_doc`, `_score`) with search after paging.** These keys are only > stable *within* a point-in-time. In the default `Live` mode the underlying `search_after` cursor > can be invalidated by index refreshes or segment merges, causing pagination to **silently skip > documents or stop early** — this is especially likely when you write to the same index you are > paging over (read-modify-write). See Elasticsearch's own > [paginate search results](https://www.elastic.co/docs/reference/elasticsearch/rest-apis/paginate-search-results#search-after) > guide for details on why `_doc` is only safe within a point-in-time. The repository logs a > warning when it detects this. Sort by a stable, unique field, or open a point-in-time snapshot: > > ```csharp > // Frozen view: _doc/_score stay stable and the cursor remains valid across pages. > var results = await repository.FindAsync( > q => q.SortDescending(e => e.CreatedUtc), > o => o.SearchAfterPaging(SearchAfterPagingMode.PointInTime)); > ``` > > Note: the repository always appends the document id as a tiebreaker, so a query with no explicit > sort is safe. The danger is an explicit unstable sort key in `Live` mode. ## Aggregations ### Aggregation Expression ```csharp var results = await repository.CountAsync(q => q .AggregationsExpression("terms:department terms:status")); // Access aggregation results var departmentAgg = results.Aggregations.Terms("terms_department"); foreach (var bucket in departmentAgg.Buckets) { Console.WriteLine($"{bucket.Key}: {bucket.Total}"); } ``` ### Common Aggregations ```csharp // Terms aggregation var results = await repository.CountAsync(q => q .AggregationsExpression("terms:department")); // Date histogram var results = await repository.CountAsync(q => q .AggregationsExpression("date:createdUtc")); // Cardinality (unique count) var results = await repository.CountAsync(q => q .AggregationsExpression("cardinality:userId")); // Statistics var results = await repository.CountAsync(q => q .AggregationsExpression("avg:salary min:salary max:salary")); // Multiple aggregations var results = await repository.CountAsync(q => q .AggregationsExpression("terms:department avg:salary cardinality:userId")); ``` ### Accessing Aggregation Results ```csharp var results = await repository.CountAsync(q => q .AggregationsExpression("terms:department avg:salary max:createdUtc")); // Terms aggregation var deptAgg = results.Aggregations.Terms("terms_department"); foreach (var bucket in deptAgg.Buckets) { Console.WriteLine($"Department: {bucket.Key}, Count: {bucket.Total}"); } // Value aggregations var avgSalary = results.Aggregations.Average("avg_salary")?.Value; var maxDate = results.Aggregations.Max("max_createdUtc")?.Value; Console.WriteLine($"Average Salary: {avgSalary}"); Console.WriteLine($"Latest Created: {maxDate}"); ``` ### Nested Field Aggregations When aggregating on fields that are mapped as `nested` in Elasticsearch, the framework automatically wraps the aggregation in a nested aggregation context. Use the same `parentObject.childField` syntax you use for flat fields: ```csharp // Terms aggregation on a nested field var results = await repository.CountAsync(q => q .AggregationsExpression("terms:peerReviews.rating")); // Multiple aggregation types on nested fields var results = await repository.CountAsync(q => q .AggregationsExpression("terms:peerReviews.reviewerEmployeeId min:peerReviews.rating max:peerReviews.rating")); ``` The framework detects nested fields via the index mapping and groups all nested aggregations under a single `SingleBucketAggregate` keyed by the nested path. Access results through that wrapper: ```csharp var results = await repository.CountAsync(q => q .AggregationsExpression("terms:peerReviews.rating min:peerReviews.rating max:peerReviews.rating")); // All nested aggregations are grouped under a single-bucket aggregate var nestedAgg = results.Aggregations["nested_peerReviews"] as SingleBucketAggregate; var ratingTerms = nestedAgg.Aggregations.Terms("terms_peerReviews.rating"); foreach (var bucket in ratingTerms.Buckets) { Console.WriteLine($"Rating {bucket.Key}: {bucket.Total}"); } var minRating = nestedAgg.Aggregations.Min("min_peerReviews.rating")?.Value; var maxRating = nestedAgg.Aggregations.Max("max_peerReviews.rating")?.Value; ``` Include and exclude filtering works the same way as non-nested aggregations: ```csharp // Only include specific terms var results = await repository.CountAsync(q => q .AggregationsExpression("terms:(peerReviews.reviewerEmployeeId @include:emp1 @include:emp2)")); // Exclude specific terms var results = await repository.CountAsync(q => q .AggregationsExpression("terms:(peerReviews.rating @exclude:1 @exclude:2)")); ``` ### Top Hits Aggregation The `tophits` sub-aggregation returns the top matching documents within each bucket: ```csharp var results = await repository.CountAsync(q => q .AggregationsExpression("terms:(age tophits:_)")); var bucket = results.Aggregations.Terms("terms_age").Buckets.First(); var topHits = bucket.Aggregations.TopHits(); var employees = topHits.Documents(); ``` ::: warning TopHitsAggregate Cannot Be Serialized `TopHitsAggregate` holds `ILazyDocument` references that contain raw Elasticsearch document bytes and require an active serializer instance to materialize into typed objects. These references are lost during JSON serialization, which means: * **Caching**: `CountResult` or `FindResults` containing `TopHitsAggregate` cannot be cached and restored via JSON serialization (Newtonsoft or System.Text.Json). The top hits data will be `null` after deserialization. * **Workaround**: If you need to cache results that include top hits, materialize the documents into concrete types *before* caching, and cache those typed results separately. ::: ## Nested Queries When querying fields inside [nested objects](https://www.elastic.co/guide/en/elasticsearch/reference/current/nested.html), the framework automatically wraps filter expressions in the required Elasticsearch `nested` query. You do not need to manually construct nested queries -- just use dotted field paths. ### Prerequisites The field must be mapped as `nested` in your index configuration: ```csharp public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.Id) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) .Nested(e => e.PeerReviews, n => n.Properties(p1 => p1 .Keyword("reviewerEmployeeId") .IntegerNumber("rating"))) ); } ``` ### Basic Nested Queries Query nested fields with standard filter expressions using `parentObject.childField` syntax: ```csharp // Exact match on a nested field var results = await repository.FindAsync(q => q .FilterExpression("peerReviews.rating:5")); // Range query on a nested field var results = await repository.FindAsync(q => q .FilterExpression("peerReviews.rating:[4 TO 5]")); // Match on a nested keyword field var results = await repository.FindAsync(q => q .FilterExpression("peerReviews.reviewerEmployeeId:bob_456")); ``` ### Combining Nested Conditions Multiple conditions on the same nested path are combined into a single `nested` query with a `bool` clause: ```csharp // AND: both conditions must match the SAME nested document var results = await repository.FindAsync(q => q .FilterExpression("peerReviews.rating:5 AND peerReviews.reviewerEmployeeId:bob_456")); // OR: either condition can match across different nested documents var results = await repository.FindAsync(q => q .FilterExpression("peerReviews.rating:>=4 OR peerReviews.reviewerEmployeeId:bob_456")); ``` ### Mixing Nested and Non-Nested Fields Nested fields and regular fields can be used together. The framework only wraps the nested portions in a `nested` query: ```csharp // "name" is a root-level field, "peerReviews.rating" is nested var results = await repository.FindAsync(q => q .FilterExpression("name:Alice peerReviews.rating:5")); ``` ### Negating Nested Conditions ```csharp // Exclude employees who have any peer review with rating 5 var results = await repository.FindAsync(q => q .FilterExpression("NOT peerReviews.rating:5")); ``` ### Default Fields with Nested Paths When [default search fields](#default-search-fields) include nested field paths, the framework automatically wraps the corresponding portion of the multi-match query in a `nested` query. No additional configuration is needed beyond including the dotted path in `SetDefaultFields`: ```csharp protected override void ConfigureQueryParser(ElasticQueryParserConfiguration config) { base.ConfigureQueryParser(config); config.SetDefaultFields([ nameof(Employee.Id).ToLowerInvariant(), nameof(Employee.Name).ToLowerInvariant(), "peerReviews.reviewerEmployeeId" // nested field ]); } ``` With this configuration, a bare search term like `bob_456` will match against `id` and `name` (root-level) as well as `peerReviews.reviewerEmployeeId` (nested). The nested portion is automatically detected via the index mapping: ```csharp // Searches id, name, AND the nested peerReviews.reviewerEmployeeId field var results = await repository.FindAsync(q => q.SearchExpression("bob_456")); ``` ### Sorting on Nested Fields Sort expressions on nested fields automatically include the required `nested` context: ```csharp // Sort descending by a nested numeric field var results = await repository.FindAsync(q => q .SortExpression("-peerReviews.rating")); ``` ### Exists / Missing on Nested Fields `_exists_` and `_missing_` queries on nested fields are automatically wrapped in a `nested` query: ```csharp // Find employees that have at least one peer review with a reviewerEmployeeId var results = await repository.FindAsync(q => q .FilterExpression("_exists_:peerReviews.reviewerEmployeeId")); ``` ### Deeply Nested Types Multi-level nesting (nested objects inside other nested objects) is supported. The framework resolves the correct nested path at each level: ```csharp // Given a mapping: parent (nested) -> child (nested inside parent) // Query a deeply nested field var results = await repository.FindAsync(q => q .FilterExpression("parent.child.field1:value")); ``` ### Known Limitations | Limitation | Details | |---|---| | **TopHits round-tripping** | `TopHitsAggregate` cannot survive JSON serialization. See the [Top Hits Aggregation](#top-hits-aggregation) warning above. | ## Field Selection Field selection controls which fields are returned from Elasticsearch via `_source` filtering. This reduces network payload and deserialization cost when you only need a subset of fields from a document. ### Including Fields Use `.Include()` to specify individual fields to return: ```csharp var results = await repository.FindAsync( query, o => o.Include(e => e.Id).Include(e => e.Name).Include(e => e.Email)); ``` You can also pass multiple fields at once: ```csharp var results = await repository.FindAsync( query, o => o.Include(e => e.Id, e => e.Name, e => e.Email)); ``` ### Excluding Fields Use `.Exclude()` to omit specific fields while returning everything else: ```csharp var results = await repository.FindAsync( query, o => o.Exclude(e => e.LargeContent).Exclude(e => e.Attachments)); ``` ### Field Mask Expressions For complex field selections, use `.IncludeMask()` or `.ExcludeMask()` with a Google FieldMask-style expression. Nested fields are grouped with parentheses and comma-separated: | Expression | Expanded Fields | |---|---| | `"id,name"` | `id`, `name` | | `"address(street,city)"` | `address.street`, `address.city` | | `"results(id,program(name,id))"` | `results.id`, `results.program.name`, `results.program.id` | ```csharp var results = await repository.FindAsync( query, o => o.IncludeMask("id,name,address(street,city,state)")); ``` Masks and individual `.Include()`/`.Exclude()` calls are additive -- they are merged into a single set at query time. ### Query-Level vs Options-Level Field includes and excludes can be set on both the query and the command options. Both sources are merged at execution time: ```csharp var results = await repository.FindAsync( q => q.Include(e => e.Name), o => o.Include(e => e.Email)); // Both Name and Email are included ``` This is useful when a repository method sets default options-level field restrictions while callers add query-level overrides. ### Merge and Precedence Rules At execution time, the repository merges all field selection settings: 1. **All includes are merged**: individual fields from `.Include()` and parsed fields from `.IncludeMask()` from both the query and command options combine into one include set. 2. **All excludes are merged**: same for `.Exclude()` and `.ExcludeMask()`. 3. **Includes win over excludes**: if the same field appears in both includes and excludes, it is included (the exclude is dropped). 4. **Automatic `Id` field**: when any includes are specified on an entity that implements `IIdentity`, the `Id` field is automatically added to ensure the document identity is always available. ### Default Excludes Repositories can register fields to exclude by default by calling `AddDefaultExclude()` in the constructor: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(/* ... */) { AddDefaultExclude(e => e.InternalNotes); AddDefaultExclude(e => e.AuditLog); } } ``` Default excludes are only applied when **no explicit excludes** are set on the query. As soon as the caller specifies any `.Exclude()` call, the defaults are skipped entirely. This prevents unexpected interactions between default and explicit excludes. ### Required Fields Repositories can register fields that must always be present when any caller-specified `_source` field restrictions are active. This is useful for: * **Multi-tenancy**: Ensuring `OrganizationId` or `ProjectId` is always available for authorization checks * **Cache invalidation**: Fields used as custom cache keys must be present to invalidate correctly * **Event handling**: Fields needed by `DocumentsChanged` / `DocumentsChanging` handlers Call `AddRequiredField()` in the repository constructor. Multiple fields can be registered in a single call: ```csharp public class StackRepository : ElasticRepositoryBase { public StackRepository(MyAppElasticConfiguration configuration) : base(configuration.Stacks) { AddRequiredField(s => s.OrganizationId, s => s.ProjectId); } } ``` #### When Required Fields Are Injected Required fields are **only injected when field restrictions are active** — i.e., when any includes, excludes, masks, or default excludes are present. When no restrictions exist at all, the full `_source` is returned and required fields have no effect. When repository-internal `AddDefaultExclude()` registrations are the only excludes present, required field injection runs but is effectively a no-op — default excludes never overlap with required fields like `Id` or `CreatedUtc`, so no fields are added or removed. If a field is registered as both a required field and a default exclude, the required field takes precedence — the field will be removed from the exclude set. #### How Required Fields Are Applied The behavior depends on what the caller specified: * **Caller has includes** (with or without excludes): Required fields are added to the include set. This ensures they appear in the narrowed result alongside the caller's selected fields. * **Caller has only excludes**: Required fields are removed from the exclude set. This preserves the "return everything except X" semantics while ensuring required fields cannot be excluded. All other non-excluded fields remain present. #### Precedence Rules When the caller specifies both includes and excludes, and a required field appears in both sets, the **include takes precedence** — the field is returned. This mirrors how systems like OData and GraphQL handle mandatory fields in projections: identity and authorization fields are always present regardless of what the client requests. #### Affected Operations Required fields apply to all source-filtered operations: `GetByIdAsync`, `GetByIdsAsync`, `FindAsync`, `PatchAllAsync`, and `BatchProcessAsync`. #### Impact on Minimal-Field Queries When you use `AddRequiredField`, callers requesting minimal fields (e.g., `.Include(e => e.Id)` for a lightweight list view) will also receive the required fields. Factor this into your API design — required fields add a small payload overhead to every partial-document response but ensure correctness for authorization, caching, and event handling. #### Custom Cache Key Fields If your repository uses custom cache keys based on specific field values (e.g., caching by `CompanyId`), register those fields as required to ensure cache invalidation works correctly even when callers request partial documents: ```csharp public class EventRepository : ElasticRepositoryBase { public EventRepository(MyAppElasticConfiguration configuration) : base(configuration.Events) { AddRequiredField(e => e.OrganizationId); } protected override async Task InvalidateCacheAsync( IReadOnlyCollection> documents, ChangeType? changeType = null) { await base.InvalidateCacheAsync(documents, changeType); // OrganizationId is guaranteed present because it's a required field await Cache.RemoveAllAsync(documents.Select(d => $"org:{d.Value.OrganizationId}")); } } ``` ### Caching Impact When includes or excludes are active, the repository skips ID-based caching to avoid storing incomplete documents in the cache. This means: * `GetByIdAsync` / `GetByIdsAsync` with field restrictions will always hit Elasticsearch directly. * Queries with custom cache keys still function normally since they cache the complete filtered result as-is. If performance is important and you frequently fetch partial documents, consider using a dedicated query with a custom cache key rather than relying on ID-based caching. ## Count and Exists ### Count with Query ```csharp var count = await repository.CountAsync(q => q.FieldEquals(e => e.Department, "Engineering")); Console.WriteLine($"Engineering employees: {count.Total}"); ``` ### Exists with Query ```csharp bool hasActiveEmployees = await repository.ExistsAsync( q => q.FieldEquals(e => e.Status, "active")); ``` ## Building Complex Queries ### Combining Query Methods ```csharp var results = await repository.FindAsync(q => q .FieldEquals(e => e.Status, "active") .FieldEquals(e => e.Department, "Engineering") .DateRange(DateTime.UtcNow.AddYears(-1), DateTime.UtcNow, e => e.HireDate) .SortExpression("-salary") .AggregationsExpression("terms:title avg:salary"), o => o.PageLimit(50)); ``` ### Reusable Query Objects ```csharp var query = new RepositoryQuery() .FieldEquals(e => e.Department, "Engineering") .FieldCondition(e => e.Name, ComparisonOperator.Contains, "John"); var results = await repository.FindAsync(q => query); var count = await repository.CountAsync(q => query); ``` ### Custom Query Extensions Create domain-specific query methods: ```csharp public static class EmployeeQueryExtensions { public static IRepositoryQuery ActiveInDepartment( this IRepositoryQuery query, string department) { return query .FieldEquals(e => e.Status, "active") .FieldEquals(e => e.Department, department); } public static IRepositoryQuery HiredBetween( this IRepositoryQuery query, DateTime start, DateTime end) { return query.DateRange(start, end, e => e.HireDate); } } // Usage var results = await repository.FindAsync(q => q .ActiveInDepartment("Engineering") .HiredBetween(DateTime.UtcNow.AddYears(-2), DateTime.UtcNow)); ``` ## Query Logging Enable query logging for debugging: ```csharp var results = await repository.FindAsync( query, o => o.QueryLogLevel(Microsoft.Extensions.Logging.LogLevel.Debug)); ``` ## Async Queries For long-running queries: ```csharp // Start async query var results = await repository.FindAsync( query, o => o.AsyncQuery(waitTime: TimeSpan.FromSeconds(5), ttl: TimeSpan.FromHours(1))); if (results.Total == 0 && results.IsAsyncQueryRunning()) { // Query is still running, get the ID var queryId = results.GetAsyncQueryId(); // Check later var laterResults = await repository.FindAsync( query, o => o.AsyncQueryId(queryId, waitTime: TimeSpan.FromSeconds(30))); } ``` ## Next Steps * [Configuration](/guide/configuration) - Query configuration options * [Caching](/guide/caching) - Cache query results * [Soft Deletes](/guide/soft-deletes) - Query soft-deleted documents * [Index Management](/guide/index-management) - Query across multiple indexes --- --- url: /guide/configuration.md --- # Configuration Options This guide covers all configuration options available in Foundatio.Repositories, including repository-level settings and per-operation options. ## Repository-Level Configuration These settings are configured in your repository constructor and apply to all operations by default. ### DefaultConsistency Controls the default refresh behavior for write operations: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { DefaultConsistency = Consistency.Immediate; } } ``` | Value | Description | |-------|-------------| | `Consistency.Eventual` | No refresh after write (default, fastest) | | `Consistency.Immediate` | Refresh immediately after write | | `Consistency.Wait` | Wait for refresh to complete | ### DefaultCacheExpiration Default cache TTL for cached operations: ```csharp DefaultCacheExpiration = TimeSpan.FromMinutes(10); ``` Default: 5 minutes ### DefaultPageLimit / MaxPageLimit Pagination limits: ```csharp DefaultPageLimit = 25; // Default page size MaxPageLimit = 1000; // Maximum allowed page size ``` Defaults: 10 / 10000 ### NotificationsEnabled Enable/disable entity change notifications via message bus: ```csharp NotificationsEnabled = true; ``` Default: `true` if a message bus is configured ### OriginalsEnabled Track original document state during save operations for change detection: ```csharp OriginalsEnabled = true; ``` When enabled: * Original document is fetched before save * Enables soft delete transition detection (`IsDeleted: false → true` sends `ChangeType.Removed`) * Enables change tracking in `DocumentsSaving`/`DocumentsSaved` events Default: `false` ### BatchNotifications Batch multiple notifications together: ```csharp BatchNotifications = true; ``` Default: `false` ### NotificationDeliveryDelay Delay notification delivery to allow Elasticsearch indexing to complete: ```csharp NotificationDeliveryDelay = TimeSpan.FromSeconds(2); ``` ::: warning Only set a delay if your message bus implementation supports delayed delivery. Message buses that don't support delayed delivery may silently drop messages. ::: Default: `null` (immediate delivery) ### DefaultPipeline Elasticsearch ingest pipeline for document processing: ```csharp DefaultPipeline = "my-ingest-pipeline"; ``` Default: `null` ### AutoCreateCustomFields Automatically create custom field definitions for unmapped fields: ```csharp AutoCreateCustomFields = true; ``` Default: `false` ### DefaultQueryLogLevel Log level for query logging: ```csharp DefaultQueryLogLevel = LogLevel.Debug; ``` Default: `LogLevel.Trace` ### Complete Repository Configuration Example ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { // Consistency DefaultConsistency = Consistency.Immediate; // Caching DefaultCacheExpiration = TimeSpan.FromMinutes(10); // Pagination DefaultPageLimit = 25; MaxPageLimit = 1000; // Notifications NotificationsEnabled = true; BatchNotifications = false; NotificationDeliveryDelay = TimeSpan.FromSeconds(1); // Change tracking OriginalsEnabled = true; // Elasticsearch DefaultPipeline = null; AutoCreateCustomFields = false; // Logging DefaultQueryLogLevel = LogLevel.Debug; } } ``` ## Per-Operation Options Override default settings for specific operations using `ICommandOptions`. ### Consistency Options ```csharp // Set consistency mode await repository.AddAsync(entity, o => o.Consistency(Consistency.Immediate)); // Shorthand for immediate consistency await repository.AddAsync(entity, o => o.ImmediateConsistency()); // Wait for refresh await repository.AddAsync(entity, o => o.ImmediateConsistency(shouldWait: true)); ``` ### Cache Options ```csharp // Enable caching await repository.GetByIdAsync(id, o => o.Cache()); // Enable with specific key await repository.FindOneAsync(query, o => o.Cache("my-cache-key")); // Enable with key and expiration await repository.FindOneAsync(query, o => o.Cache("my-key", TimeSpan.FromMinutes(5))); // Set cache key separately await repository.FindAsync(query, o => o.CacheKey("employees-active")); // Set expiration await repository.GetByIdAsync(id, o => o.CacheExpiresIn(TimeSpan.FromMinutes(30))); await repository.GetByIdAsync(id, o => o.CacheExpiresAt(DateTime.UtcNow.AddHours(1))); // Read from cache only (don't write) await repository.GetByIdAsync(id, o => o.ReadCache()); // Disable caching for this operation await repository.GetByIdAsync(id, o => o.Cache(false)); ``` ### Validation Options ```csharp // Skip validation await repository.AddAsync(entity, o => o.SkipValidation()); // Explicitly control validation await repository.SaveAsync(entity, o => o.Validation(false)); ``` ### Notification Options ```csharp // Disable notifications for this operation await repository.AddAsync(entity, o => o.Notifications(false)); // Enable notifications (override if disabled at repository level) await repository.SaveAsync(entity, o => o.Notifications(true)); ``` ### Pagination Options ```csharp // Set page number and limit await repository.FindAsync(query, o => o.PageNumber(2).PageLimit(50)); // Snapshot paging (scroll API) await repository.FindAsync(query, o => o.SnapshotPaging()); await repository.FindAsync(query, o => o.SnapshotPagingLifetime(TimeSpan.FromMinutes(5))); // Search-after paging await repository.FindAsync(query, o => o.SearchAfterPaging()); await repository.FindAsync(query, o => o.SearchAfterToken("token")); ``` ### Soft Delete Options ```csharp // Include soft-deleted documents await repository.FindAsync(query, o => o.IncludeSoftDeletes()); // Set soft delete mode await repository.FindAsync(query, o => o.SoftDeleteMode(SoftDeleteQueryMode.All)); await repository.FindAsync(query, o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); await repository.FindAsync(query, o => o.SoftDeleteMode(SoftDeleteQueryMode.ActiveOnly)); ``` ### Version Options ```csharp // Skip optimistic concurrency check await repository.SaveAsync(entity, o => o.SkipVersionCheck()); // Explicitly control version checking await repository.SaveAsync(entity, o => o.VersionCheck(false)); ``` ### Originals Options ```csharp // Enable original document tracking for this operation await repository.SaveAsync(entity, o => o.Originals(true)); // Provide original documents manually await repository.SaveAsync(entity, o => o.AddOriginals(originalEntity)); await repository.SaveAsync(entities, o => o.AddOriginals(originalEntities)); ``` ### Field Selection Options ```csharp // Include specific fields await repository.FindAsync(query, o => o .Include(e => e.Id) .Include(e => e.Name) .Include(e => e.Email)); // Include using mask pattern await repository.FindAsync(query, o => o.IncludeMask("id,name,email,address.*")); // Exclude specific fields await repository.FindAsync(query, o => o .Exclude(e => e.LargeContent) .Exclude(e => e.InternalData)); // Exclude using mask pattern await repository.FindAsync(query, o => o.ExcludeMask("largeContent,internal*")); ``` ### Timeout and Retry Options ```csharp // Set query timeout await repository.FindAsync(query, o => o.Timeout(TimeSpan.FromSeconds(30))); // Set retry count await repository.SaveAsync(entity, o => o.Retry(5)); ``` ### Query Logging Options ```csharp // Set log level for this query await repository.FindAsync(query, o => o.QueryLogLevel(LogLevel.Debug)); ``` ### Async Query Options ```csharp // Enable async query await repository.FindAsync(query, o => o.AsyncQuery()); await repository.FindAsync(query, o => o.AsyncQuery( waitTime: TimeSpan.FromSeconds(5), ttl: TimeSpan.FromHours(1))); // Get async query results by ID await repository.FindAsync(query, o => o.AsyncQueryId( "query-id-123", waitTime: TimeSpan.FromSeconds(30), autoDelete: true)); ``` ### Combining Options ```csharp await repository.FindAsync( q => q.FilterExpression("status:active"), o => o .ImmediateConsistency() .Cache("active-employees", TimeSpan.FromMinutes(5)) .PageLimit(100) .Include(e => e.Id) .Include(e => e.Name) .QueryLogLevel(LogLevel.Debug)); ``` ## ConfigureOptions Override Override `ConfigureOptions` to set custom defaults for all operations: ```csharp public class EmployeeRepository : ElasticRepositoryBase { private readonly string _tenantId; public EmployeeRepository(EmployeeIndex index, ITenantContext tenant) : base(index) { _tenantId = tenant.TenantId; } protected override ICommandOptions ConfigureOptions(ICommandOptions options) { options = base.ConfigureOptions(options); // Add custom defaults options.DefaultCacheKey($"tenant:{_tenantId}"); return options; } } ``` ## Configuration Summary Table | Setting | Default | Repository Level | Per-Operation | |---------|---------|------------------|---------------| | Consistency | `Eventual` | `DefaultConsistency` | `.Consistency()`, `.ImmediateConsistency()` | | Cache Expiration | 5 minutes | `DefaultCacheExpiration` | `.CacheExpiresIn()` | | Page Limit | 10 | `DefaultPageLimit` | `.PageLimit()` | | Max Page Limit | 10000 | `MaxPageLimit` | N/A | | Notifications | true (if bus) | `NotificationsEnabled` | `.Notifications()` | | Originals | false | `OriginalsEnabled` | `.Originals()`, `.AddOriginals()` | | Batch Notifications | false | `BatchNotifications` | N/A | | Notification Delay | null | `NotificationDeliveryDelay` | N/A | | Pipeline | null | `DefaultPipeline` | N/A | | Auto Custom Fields | false | `AutoCreateCustomFields` | N/A | | Query Log Level | Trace | `DefaultQueryLogLevel` | `.QueryLogLevel()` | | Validation | true | N/A | `.SkipValidation()`, `.Validation()` | | Soft Deletes | ActiveOnly | N/A | `.IncludeSoftDeletes()`, `.SoftDeleteMode()` | | Version Check | true | N/A | `.SkipVersionCheck()`, `.VersionCheck()` | ## Next Steps * [Validation](/guide/validation) - Document validation * [Caching](/guide/caching) - Cache configuration details * [Message Bus](/guide/message-bus) - Notification configuration * [Soft Deletes](/guide/soft-deletes) - Soft delete configuration --- --- url: /guide/validation.md --- # Validation Foundatio.Repositories provides a validation system for ensuring document integrity before persistence. This guide covers implementing validation, handling exceptions, and validation patterns. ## Document Validation ### ValidateAndThrowAsync Override `ValidateAndThrowAsync` in your repository to implement custom validation: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { } protected override Task ValidateAndThrowAsync(Employee document) { if (string.IsNullOrEmpty(document.Name)) throw new DocumentValidationException("Name is required"); if (string.IsNullOrEmpty(document.Email)) throw new DocumentValidationException("Email is required"); if (!IsValidEmail(document.Email)) throw new DocumentValidationException("Email format is invalid"); if (document.Age < 0 || document.Age > 150) throw new DocumentValidationException("Age must be between 0 and 150"); return Task.CompletedTask; } private bool IsValidEmail(string email) { return email.Contains('@') && email.Contains('.'); } } ``` ### When Validation Runs Validation is called automatically during: * `AddAsync` - Before adding new documents * `SaveAsync` - Before saving/updating documents Validation is **not** called during: * `PatchAsync` / `PatchAllAsync` - Partial updates bypass validation * `RemoveAsync` / `RemoveAllAsync` - Deletions don't require validation ### Async Validation For validation that requires async operations (e.g., checking uniqueness): ```csharp protected override async Task ValidateAndThrowAsync(Employee document) { // Basic validation if (string.IsNullOrEmpty(document.Email)) throw new DocumentValidationException("Email is required"); // Check for duplicate email var existing = await FindOneAsync(q => q .FieldEquals(e => e.Email, document.Email) .ExcludedId(document.Id)); if (existing != null) throw new DuplicateDocumentException($"Email {document.Email} is already in use"); } ``` ## Exception Hierarchy Foundatio.Repositories provides a hierarchy of exceptions for different error scenarios: ```mermaid classDiagram Exception <|-- RepositoryException RepositoryException <|-- DocumentException DocumentException <|-- DocumentValidationException DocumentException <|-- DocumentNotFoundException DocumentException <|-- DuplicateDocumentException DocumentException <|-- VersionConflictDocumentException class DocumentNotFoundException { +string Id } ``` ### DocumentValidationException Thrown when document validation fails: ```csharp public class DocumentValidationException : DocumentException { public DocumentValidationException() { } public DocumentValidationException(string message) : base(message) { } } ``` **Usage:** ```csharp throw new DocumentValidationException("Name is required"); throw new DocumentValidationException($"Age {age} is out of valid range"); ``` ### DocumentNotFoundException Thrown when a document is not found: ```csharp public class DocumentNotFoundException : DocumentException { public string Id { get; } public DocumentNotFoundException(string id) : base($"Document \"{id}\" could not be found") { Id = id; } } ``` **When thrown:** * `PatchAsync` when document doesn't exist * `SaveAsync` when updating a non-existent document (in some cases) * `RemoveAsync` when deleting a non-existent document ### DuplicateDocumentException Thrown when attempting to create a duplicate document: ```csharp public class DuplicateDocumentException : DocumentException { public DuplicateDocumentException(string message) : base(message) { } } ``` **Usage:** ```csharp throw new DuplicateDocumentException($"Document with email {email} already exists"); ``` ### VersionConflictDocumentException Thrown when optimistic concurrency check fails: ```csharp public class VersionConflictDocumentException : DocumentException { public VersionConflictDocumentException() { } public VersionConflictDocumentException(string message) : base(message) { } public VersionConflictDocumentException(string message, Exception inner) : base(message, inner) { } } ``` **When thrown:** * `SaveAsync` when the document version doesn't match (if `IVersioned`) ## Skipping Validation ### Per-Operation Skip validation for specific operations: ```csharp // Skip validation for trusted data await repository.AddAsync(trustedEntity, o => o.SkipValidation()); // Explicitly control validation await repository.SaveAsync(entity, o => o.Validation(false)); ``` ### Use Cases for Skipping Validation * **Bulk imports** - Trusted data from verified sources * **System operations** - Internal updates that bypass business rules * **Migrations** - Data transformations during schema changes ::: warning Only skip validation when you're certain the data is valid. Invalid data can cause issues downstream. ::: ## Validation Patterns ### Fluent Validation Integration Integrate with FluentValidation for complex validation rules: ```csharp public class EmployeeValidator : AbstractValidator { public EmployeeValidator() { RuleFor(e => e.Name) .NotEmpty().WithMessage("Name is required") .MaximumLength(100).WithMessage("Name cannot exceed 100 characters"); RuleFor(e => e.Email) .NotEmpty().WithMessage("Email is required") .EmailAddress().WithMessage("Invalid email format"); RuleFor(e => e.Age) .InclusiveBetween(18, 100).WithMessage("Age must be between 18 and 100"); } } public class EmployeeRepository : ElasticRepositoryBase { private readonly IValidator _validator; public EmployeeRepository(EmployeeIndex index, IValidator validator) : base(index) { _validator = validator; } protected override async Task ValidateAndThrowAsync(Employee document) { var result = await _validator.ValidateAsync(document); if (!result.IsValid) { var errors = string.Join("; ", result.Errors.Select(e => e.ErrorMessage)); throw new DocumentValidationException(errors); } } } ``` ### Validation in Event Handlers Validate in event handlers for cross-cutting concerns: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { DocumentsAdding.AddHandler(ValidateNewEmployee); DocumentsSaving.AddHandler(ValidateEmployeeUpdate); } private Task ValidateNewEmployee(object sender, DocumentsEventArgs args) { foreach (var employee in args.Documents) { // Validate new employee specific rules if (employee.StartDate < DateTime.UtcNow.Date) throw new DocumentValidationException("Start date cannot be in the past"); } return Task.CompletedTask; } private Task ValidateEmployeeUpdate(object sender, ModifiedDocumentsEventArgs args) { foreach (var modified in args.Documents) { var original = modified.Original; var current = modified.Value; // Prevent certain changes if (original != null && original.EmployeeId != current.EmployeeId) throw new DocumentValidationException("Employee ID cannot be changed"); } return Task.CompletedTask; } } ``` ### Conditional Validation Apply different validation rules based on context: ```csharp protected override Task ValidateAndThrowAsync(Employee document) { // Always validate required fields if (string.IsNullOrEmpty(document.Name)) throw new DocumentValidationException("Name is required"); // Additional validation for active employees if (document.Status == EmployeeStatus.Active) { if (string.IsNullOrEmpty(document.Department)) throw new DocumentValidationException("Active employees must have a department"); if (document.ManagerId == null) throw new DocumentValidationException("Active employees must have a manager"); } return Task.CompletedTask; } ``` ## Error Handling ### Catching Validation Errors ```csharp try { await repository.AddAsync(employee); } catch (DocumentValidationException ex) { _logger.LogWarning("Validation failed: {Message}", ex.Message); return BadRequest(ex.Message); } catch (DuplicateDocumentException ex) { _logger.LogWarning("Duplicate document: {Message}", ex.Message); return Conflict(ex.Message); } ``` ### Comprehensive Error Handling ```csharp try { await repository.SaveAsync(employee); } catch (DocumentValidationException ex) { // Validation failed return BadRequest(new { error = "Validation failed", message = ex.Message }); } catch (DocumentNotFoundException ex) { // Document doesn't exist return NotFound(new { error = "Not found", id = ex.Id }); } catch (VersionConflictDocumentException ex) { // Concurrent modification return Conflict(new { error = "Version conflict", message = ex.Message }); } catch (DuplicateDocumentException ex) { // Duplicate document return Conflict(new { error = "Duplicate", message = ex.Message }); } catch (RepositoryException ex) { // Other repository errors _logger.LogError(ex, "Repository error"); return StatusCode(500, new { error = "Internal error" }); } ``` ### Validation Result Pattern For APIs that need to return validation errors without exceptions: ```csharp public class ValidationResult { public bool IsValid { get; set; } public List Errors { get; set; } = new(); } public class EmployeeService { private readonly IEmployeeRepository _repository; public async Task<(Employee Employee, ValidationResult Validation)> CreateEmployeeAsync( Employee employee) { var validation = ValidateEmployee(employee); if (!validation.IsValid) return (null, validation); try { var result = await _repository.AddAsync(employee); return (result, validation); } catch (DocumentValidationException ex) { validation.IsValid = false; validation.Errors.Add(ex.Message); return (null, validation); } } private ValidationResult ValidateEmployee(Employee employee) { var result = new ValidationResult { IsValid = true }; if (string.IsNullOrEmpty(employee.Name)) result.Errors.Add("Name is required"); if (string.IsNullOrEmpty(employee.Email)) result.Errors.Add("Email is required"); result.IsValid = result.Errors.Count == 0; return result; } } ``` ## Next Steps * [Configuration](/guide/configuration) - Validation configuration options * [Repository Pattern](/guide/repository-pattern) - Event handlers for validation * [CRUD Operations](/guide/crud-operations) - Error handling in operations --- --- url: /guide/caching.md --- # Caching Foundatio.Repositories provides built-in distributed caching with automatic invalidation. This guide covers cache configuration, behavior, and important gaps to be aware of. ## Overview Caching in Foundatio.Repositories is built on Foundatio's `ICacheClient` abstraction, supporting: * In-memory caching (development/testing) * Redis (distributed) * Hybrid (in-memory L1 + distributed L2 for reduced network round trips) * Any custom `ICacheClient` implementation ## Configuration ### Enable Caching Provide an `ICacheClient` to your Elasticsearch configuration: ```csharp using Foundatio.Caching; public class MyElasticConfiguration : ElasticConfiguration { public MyElasticConfiguration(ICacheClient cache, ILoggerFactory loggerFactory) : base(cache: cache, loggerFactory: loggerFactory) { AddIndex(Employees = new EmployeeIndex(this)); } public EmployeeIndex Employees { get; } } ``` ### Cache Implementations ```csharp // In-memory (for development/testing) services.AddSingleton(new InMemoryCacheClient()); // Redis services.AddSingleton(sp => new RedisCacheClient(o => o.ConnectionMultiplexer(sp.GetRequiredService()))); // Hybrid (in-memory L1 + distributed L2 - best for production, reduces network round trips) services.AddSingleton(sp => new RedisHybridCacheClient(o => o.ConnectionMultiplexer(sp.GetRequiredService()))); ``` ### Repository Cache Settings ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { // Default cache expiration (default: 5 minutes) DefaultCacheExpiration = TimeSpan.FromMinutes(10); } } ``` ### Override Cache Client Per Repository You can override the cache client for a specific repository by calling `SetCacheClient` in the constructor. This is useful when you need a different cache implementation (e.g., a hybrid cache) for a particular repository: ```csharp public class MyRepository : ElasticRepositoryBase { public MyRepository( MyElasticConfiguration elasticConfig, [FromKeyedServices("hybrid")] ICacheClient cacheClient ) : base(elasticConfig.MyIndex) { SetCacheClient(cacheClient); } } ``` You can also disable caching entirely for a repository: ```csharp public class UncachedRepository : ElasticRepositoryBase { public UncachedRepository(MyElasticConfiguration elasticConfig) : base(elasticConfig.MyIndex) { DisableCache(); } } ``` ## Using the Cache ### Cache on Read ```csharp // Cache by document ID var employee = await repository.GetByIdAsync(id, o => o.Cache()); // Cache with custom key var hit = await repository.FindOneAsync( q => q.FieldEquals(e => e.Email, email), o => o.Cache($"employee:email:{email}")); var employee = hit?.Document; // Cache with custom expiration var employee = await repository.GetByIdAsync(id, o => o.Cache().CacheExpiresIn(TimeSpan.FromMinutes(30))); ``` ### Cache Options ```csharp // Enable caching o.Cache() // Enable with specific key o.Cache("my-cache-key") // Enable with key and expiration o.Cache("my-key", TimeSpan.FromMinutes(5)) // Set cache key separately o.CacheKey("my-key") // Set expiration o.CacheExpiresIn(TimeSpan.FromMinutes(10)) o.CacheExpiresAt(DateTime.UtcNow.AddHours(1)) // Read from cache only (don't write) o.ReadCache() // Disable caching for this operation o.Cache(false) ``` ## Automatic Cache Invalidation The repository automatically invalidates cache in most scenarios. ### When Cache IS Automatically Invalidated | Operation | Behavior | |-----------|----------| | `AddAsync` | Documents added to cache after successful add | | `SaveAsync` | Cache invalidated by ID, then documents re-added | | `RemoveAsync` | Cache invalidated by document ID | | `PatchAsync` (single ID) | Cache invalidated by ID | | `PatchAsync` (multiple IDs) | Cache invalidated for all IDs | ### Code Flow for Save ```csharp // When you call SaveAsync: await repository.SaveAsync(employee); // Internally: // 1. Document is indexed to Elasticsearch // 2. Cache is invalidated for the document ID // 3. Document is added back to cache with updated values // 4. EntityChanged notification is published ``` ## Cache Behavior on Partial Failure When a bulk write operation partially fails (some documents succeed, others fail), the cache is updated only for successful documents: | Document Status | Cache Action | |-----------------|--------------| | Succeeded | Added to cache (freshest data available) | | Failed (any error) | Cache entry **unchanged** (failed writes don't mutate Elasticsearch) | Failed writes (409 conflicts, 429/503 rate limits, or other errors) do not mutate the document in Elasticsearch, so the existing cache entry (if any) remains valid. Cache consistency for concurrent writes is handled by the message bus `EntityChanged` notifications — the writer that successfully mutated the document is responsible for updating or invalidating the cache. ```csharp try { await repository.SaveAsync(documents, o => o.Cache()); } catch (VersionConflictDocumentException) { // Successful docs: cached with latest data // Conflicting docs: cache unchanged (the successful concurrent writer handles its own cache update) } ``` ## Cache Invalidation Gaps ::: warning Important There are scenarios where cache is NOT automatically invalidated. Understanding these gaps is critical for maintaining cache consistency. ::: ### PatchAllAsync - Partial Gap When using `PatchAllAsync`, only document IDs are invalidated. **Custom cache keys are NOT invalidated.** ```csharp // This will invalidate cache by document IDs await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Engineering"), new ScriptPatch("ctx._source.salary += 1000")); // But if you cached by a custom key like "dept:engineering:employees", // that cache key is NOT invalidated ``` **Solution:** Override `InvalidateCacheByQueryAsync` or manually invalidate: ```csharp // After PatchAllAsync, manually invalidate custom keys await repository.InvalidateCacheAsync("dept:engineering:employees"); ``` ### RemoveAllAsync (Query) - Conditional Gap When `RemoveAllAsync` uses `DeleteByQuery` (no event listeners and cache disabled), only query IDs are invalidated: ```csharp // If no event listeners are registered and cache is disabled, // this uses DeleteByQuery which only invalidates query IDs await repository.RemoveAllAsync(q => q.FieldEquals(e => e.Status, "inactive")); ``` **Solution:** Ensure cache is enabled or add event listeners: ```csharp // Enable cache for the operation await repository.RemoveAllAsync(query, o => o.Cache()); ``` ### Direct Elasticsearch Operations Any operations performed directly via the Elasticsearch client bypass the repository entirely: ```csharp // This bypasses the repository - NO cache invalidation await _elasticClient.IndexAsync(document, i => i.Index("employees")); ``` **Solution:** Always use repository methods for data operations. ## Manual Cache Invalidation ### Invalidate by Document ```csharp // Single document await repository.InvalidateCacheAsync(employee); // Multiple documents await repository.InvalidateCacheAsync(employees); ``` ### Invalidate by Cache Key ```csharp // Single key await repository.InvalidateCacheAsync("my-cache-key"); // Multiple keys await repository.InvalidateCacheAsync(new[] { "key1", "key2", "key3" }); ``` ### Custom Cache Key Invalidation Pattern Override `InvalidateCacheAsync` to handle custom cache keys: ```csharp public class EmployeeRepository : ElasticRepositoryBase { // Override to invalidate email cache when documents change protected override async Task InvalidateCacheAsync( IReadOnlyCollection> documents, ChangeType? changeType = null) { // Call base implementation for ID-based invalidation await base.InvalidateCacheAsync(documents, changeType); // Invalidate custom cache keys for current email addresses var emailKeys = documents .Where(d => !string.IsNullOrEmpty(d.Value.Email)) .Select(d => $"employee:email:{d.Value.Email.ToLowerInvariant()}") .ToList(); if (emailKeys.Count > 0) await Cache.RemoveAllAsync(emailKeys); // Also invalidate original email if it changed var originalEmailKeys = documents .Where(d => d.Original != null && !string.IsNullOrEmpty(d.Original.Email)) .Where(d => d.Original.Email != d.Value.Email) .Select(d => $"employee:email:{d.Original.Email.ToLowerInvariant()}") .ToList(); if (originalEmailKeys.Count > 0) await Cache.RemoveAllAsync(originalEmailKeys); } } ``` ## Soft Delete Cache Behavior For entities implementing `ISupportSoftDeletes`, the repository maintains a special cache list to handle eventual consistency. ### How It Works When a document is soft-deleted: ```csharp employee.IsDeleted = true; await repository.SaveAsync(employee); // Internally: // 1. Document ID is added to "deleted" list in cache // 2. List has 30-second TTL // 3. Queries automatically exclude IDs in the "deleted" list ``` ### Query Filtering Before queries execute, soft-deleted IDs are excluded: ```csharp // When you query: var results = await repository.FindAsync(query); // Internally (if SoftDeleteQueryMode.ActiveOnly): // 1. Check cache for "deleted" list // 2. Add excluded IDs to query // 3. Execute query ``` ### Purpose This handles the eventual consistency window where: 1. Document is soft-deleted 2. Elasticsearch hasn't indexed the change yet 3. Cache knows about the deletion 4. Queries correctly exclude the document After 30 seconds, Elasticsearch should have indexed the change, and the cache entry expires. ## Distributed Cache Consistency ### Message Bus Integration For distributed scenarios, subscribe to `EntityChanged` messages: ```csharp await messageBus.SubscribeAsync(async (msg, ct) => { if (msg.Type == nameof(Employee)) { // Invalidate local cache when other instances make changes await repository.InvalidateCacheAsync(msg.Id); } }); ``` ### NotificationDeliveryDelay Use `NotificationDeliveryDelay` to allow Elasticsearch indexing to complete before consumers read: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { NotificationDeliveryDelay = TimeSpan.FromSeconds(1); } } ``` ## Cache Best Practices ### 1. Use Consistent Cache Keys ```csharp // Good: Consistent key format o.Cache($"employee:email:{email.ToLowerInvariant()}") // Bad: Inconsistent casing o.Cache($"employee:email:{email}") // May not match invalidation ``` ### 2. Override InvalidateCacheAsync for Custom Keys ```csharp protected override async Task InvalidateCacheAsync( IReadOnlyCollection> documents, ChangeType? changeType = null) { await base.InvalidateCacheAsync(documents, changeType); // Add custom key invalidation var customKeys = documents.Select(d => $"custom:{d.Value.CustomField}"); await Cache.RemoveAllAsync(customKeys); } ``` ### 3. Be Aware of PatchAllAsync Gaps ```csharp // After bulk operations, manually invalidate if needed await repository.PatchAllAsync(query, patch); await repository.InvalidateCacheAsync("affected-cache-key"); ``` ### 4. Use Repository Methods ```csharp // Good: Uses repository, cache is managed await repository.SaveAsync(employee); // Bad: Bypasses repository, cache not invalidated await _elasticClient.IndexAsync(employee); ``` ### 5. Test Cache Behavior ```csharp // Use InMemoryCacheClient for testing var cache = new InMemoryCacheClient(); // ... perform operations ... // Check cache statistics var stats = cache.GetStats(); Console.WriteLine($"Hits: {stats.Hits}, Misses: {stats.Misses}"); ``` ## Summary: Cache Invalidation Matrix | Operation | Auto-Invalidated | Custom Keys | Notes | |-----------|------------------|-------------|-------| | `AddAsync` | Yes (adds to cache) | No | Override `InvalidateCacheAsync` | | `SaveAsync` | Yes | No | Override `InvalidateCacheAsync` | | `RemoveAsync` | Yes | No | Override `InvalidateCacheAsync` | | `PatchAsync` (ID) | Yes | No | Override `InvalidateCacheAsync` | | `PatchAllAsync` | Partial (IDs only) | No | Manual invalidation needed | | `RemoveAllAsync` | Partial | No | Use `o.Cache()` or add listeners | | Direct ES | No | No | Always use repository | ## Advanced: How Caching Handles Dirty Reads Elasticsearch's search path is eventually consistent -- see [Consistency and Dirty Reads](consistency.md) for the full explanation of which operations are real-time vs. near real-time. The cache layer is aware of this. When a search-based method like `FindOneAsync` returns results without `ImmediateConsistency`, the repository treats them as a potential dirty read and skips caching them by document ID. This prevents stale search results from polluting the ID-based cache that `GetByIdAsync` relies on: ```csharp // Internal behavior in AddDocumentsToCacheAsync: protected virtual async Task AddDocumentsToCacheAsync( ICollection> findHits, ICommandOptions options, bool isDirtyRead) { // Custom cache keys are always cached (you explicitly requested it) if (options.HasCacheKey()) { await Cache.SetAsync(options.GetCacheKey(), findHits, options.GetExpiresIn()); // ... } // Don't add dirty read documents by ID - they may be out of sync if (isDirtyRead) return; // Only cache by ID for real-time reads // ... } ``` ### Custom Cache Keys for Eventual Consistency You can use custom cache keys to make `FindOneAsync` work reliably without requiring immediate consistency: ```csharp public class UserRepository : ElasticRepositoryBase { public async Task GetByEmailAddressAsync(string emailAddress) { if (String.IsNullOrWhiteSpace(emailAddress)) return null; emailAddress = emailAddress.Trim().ToLowerInvariant(); // Use a custom cache key - this caches the result even for dirty reads var hit = await FindOneAsync( q => q.FieldEquals(u => u.EmailAddress, emailAddress), o => o.Cache(EmailCacheKey(emailAddress))); return hit?.Document; } // Override to add documents to cache by email protected override async Task AddDocumentsToCacheAsync( ICollection> findHits, ICommandOptions options, bool isDirtyRead) { await base.AddDocumentsToCacheAsync(findHits, options, isDirtyRead); // Cache by email address for future lookups var cacheEntries = new Dictionary>(); foreach (var hit in findHits.Where(d => !String.IsNullOrEmpty(d.Document?.EmailAddress))) cacheEntries.Add(EmailCacheKey(hit.Document.EmailAddress), hit); if (cacheEntries.Count > 0) await AddDocumentsToCacheWithKeyAsync(cacheEntries, options.GetExpiresIn()); } // Override to invalidate email cache when documents change protected override Task InvalidateCacheAsync( IReadOnlyCollection> documents, ChangeType? changeType = null) { // Union originals and modified values to handle field renames var keysToRemove = documents.UnionOriginalAndModified() .Where(u => !string.IsNullOrEmpty(u.EmailAddress)) .Select(u => EmailCacheKey(u.EmailAddress)) .Distinct(); return Task.WhenAll( Cache.RemoveAllAsync(keysToRemove), base.InvalidateCacheAsync(documents, changeType)); } private static string EmailCacheKey(string emailAddress) => String.Concat("Email:", emailAddress.Trim().ToLowerInvariant()); } ``` **How this works:** 1. **First call**: `FindOneAsync` searches Elasticsearch (may be a dirty read), caches result by email key 2. **Subsequent calls**: Returns cached result immediately, no Elasticsearch query needed 3. **On save/update**: `InvalidateCacheAsync` clears the email cache key 4. **Next lookup**: Fresh search, re-cached This pattern provides **eventual consistency with caching** - you don't need `Consistency.Immediate` because: * The cache is populated on first successful lookup * The cache is invalidated when the document changes * Subsequent lookups hit the cache, not Elasticsearch ## Advanced: Originals for Change Detection ### What Are Originals? When `OriginalsEnabled = true`, the repository fetches the current document from the database before saving. This allows you to: * Detect what fields changed * Access the previous values in event handlers * Properly invalidate cache for changed values (like email addresses) ### Enabling Originals ```csharp public class UserRepository : ElasticRepositoryBase { public UserRepository(UserIndex index) : base(index) { OriginalsEnabled = true; // Fetch original before save } } ``` ### Using Originals in Event Handlers ```csharp DocumentsChanging.AddHandler(async (sender, args) => { foreach (var doc in args.Documents) { if (doc.Original != null) { // Compare old and new values if (doc.Original.Email != doc.Value.Email) { _logger.LogInformation( "Email changed from {Old} to {New}", doc.Original.Email, doc.Value.Email); } } } }); ``` ### UnionOriginalAndModified Pattern A common pattern for cache invalidation is to collect keys from both the original and modified documents (to handle field changes like email address renames). The built-in `UnionOriginalAndModified` extension method (in `Foundatio.Repositories.Extensions`) combines both the current and previous versions of each document: ```csharp protected override Task InvalidateCacheAsync( IReadOnlyCollection> documents, ChangeType? changeType = null) { var keysToRemove = documents.UnionOriginalAndModified() .Where(u => !string.IsNullOrEmpty(u.EmailAddress)) .Select(u => EmailCacheKey(u.EmailAddress)) .Distinct(); return Task.WhenAll( Cache.RemoveAllAsync(keysToRemove), base.InvalidateCacheAsync(documents, changeType)); } ``` ### Per-Operation Control ```csharp // Enable originals for a specific operation await repository.SaveAsync(user, o => o.Originals(true)); // Disable originals for a specific operation (performance optimization) await repository.SaveAsync(user, o => o.Originals(false)); // Pass known originals to avoid extra database fetch await repository.SaveAsync(user, o => o.AddOriginals(originalUser)); ``` ### Performance Consideration Enabling `OriginalsEnabled` adds an extra database read before each save. Use it when you need: * Change detection in event handlers * Proper cache invalidation for non-ID fields * Soft delete notifications (to know the document was active before deletion) ## Advanced: Required Fields for Cache Invalidation ### The Problem When callers request partial documents (via `.Include()`, `.Exclude()`, or field masks), fields needed for cache invalidation or event handling may be missing. For example, if you cache by `EmailAddress` and a caller requests only `Id` and `Name`, the `InvalidateCacheAsync` override won't have the email to clear the cache. ### AddRequiredField Register fields that must always be returned when any caller-specified source filtering is active: ```csharp public class UserRepository : ElasticRepositoryBase { public UserRepository(UserIndex index) : base(index) { AddRequiredField(u => u.EmailAddress, u => u.OrganizationIds); } } ``` Required fields are automatically injected when caller-specified source filtering is active: added to the include set when includes are present, or removed from the exclude set when only excludes are present. This applies to all source-filtered operations: `GetByIdAsync`, `GetByIdsAsync`, `FindAsync`, `RemoveAllAsync`, and `PatchAllAsync`. See [Required Fields](querying.md#required-fields) for full details on injection behavior, precedence rules, and interactions with default excludes. ### Default Required Fields The repository automatically registers these as required fields: * `Id` (always required) * `CreatedUtc` (if entity implements `IHaveCreatedDate`) ### Use Cases 1. **Cache invalidation by non-ID fields**: Need the email to invalidate email cache 2. **Event handlers that need specific data**: Audit logging, notifications 3. **Cascade operations**: Need organization IDs to update related entities 4. **Multi-tenancy**: Authorization checks that need tenant identifiers ## Next Steps * [Consistency and Dirty Reads](consistency.md) - Which operations are real-time vs. eventually consistent * [Message Bus](/guide/message-bus) - Distributed cache invalidation * [Configuration](/guide/configuration) - Cache configuration options * [Soft Deletes](/guide/soft-deletes) - Soft delete cache behavior --- --- url: /guide/message-bus.md --- # Message Bus Foundatio.Repositories integrates with Foundatio's message bus to publish entity change notifications. This enables real-time updates, event-driven architectures, and distributed cache invalidation. ## Overview When documents are added, saved, or removed, the repository publishes `EntityChanged` messages to the message bus. Other parts of your system can subscribe to these messages to react to changes. ## Configuration ### Enable Message Bus Provide an `IMessageBus` to your Elasticsearch configuration: ```csharp using Foundatio.Messaging; public class MyElasticConfiguration : ElasticConfiguration { public MyElasticConfiguration( ICacheClient cache, IMessageBus messageBus, ILoggerFactory loggerFactory) : base(cache: cache, messageBus: messageBus, loggerFactory: loggerFactory) { AddIndex(Employees = new EmployeeIndex(this)); } public EmployeeIndex Employees { get; } } ``` ### Message Bus Implementations ```csharp // In-memory (for development/testing) services.AddSingleton(new InMemoryMessageBus()); // Redis services.AddSingleton(sp => new RedisMessageBus(new RedisConnection("localhost:6379"))); // RabbitMQ services.AddSingleton(sp => new RabbitMQMessageBus(new RabbitMQOptions { ConnectionString = "amqp://localhost" })); ``` ## EntityChanged Message ### Message Structure ```csharp public class EntityChanged : IHaveData { public string? Type { get; set; } // Entity type name (e.g., "Employee"); null for non-entity-specific notifications public string? Id { get; set; } // Document ID; null for bulk/type-level notifications public ChangeType ChangeType { get; set; } // Added, Saved, or Removed public IDictionary Data { get; set; } // Custom data } public enum ChangeType : byte { Added = 0, Saved = 1, Removed = 2 } ``` ### When Each ChangeType Is Used | ChangeType | Triggered By | |------------|--------------| | `Added` | `AddAsync` - New document created | | `Saved` | `SaveAsync` - Document updated | | `Saved` | `PatchAsync` - Document patched (with document ID) | | `Saved` | `PatchAsync(Ids)` - Documents patched (one message per modified ID) | | `Saved` | `PatchAllAsync` - Documents patched (per ID or type-level) | | `Removed` | `RemoveAsync` - Document deleted | | `Removed` | Soft delete transition (`IsDeleted: false → true`) via `SaveAsync` | ### Patch Notification Behavior Patch operations always use `ChangeType.Saved`. The `Id` field in the `EntityChanged` message depends on how the patch is invoked: | Method | Id Field | Notes | |--------|----------|-------| | `PatchAsync(id, ...)` | Document ID | One message per patched document | | `PatchAsync(Ids, ScriptPatch/PartialPatch)` | Document ID | One message **per modified ID** (noop IDs excluded) | | `PatchAsync(Ids, JsonPatch/ActionPatch)` | Document ID | Delegates to `PatchAllAsync` with explicit IDs — one message **per ID** in the query | | `PatchAllAsync` with explicit IDs | Document ID | One message **per ID** in the query (uncached) or **per modified ID** (cached/batch) | | `PatchAllAsync` with filter-only query (cached) | Document ID | One message **per modified ID** (sent per-batch) | | `PatchAllAsync` with filter-only query (uncached) | `null` | Single type-level notification | ::: info `PatchAllAsync` sends notifications incrementally per batch as documents are processed. Subscribers may receive messages while the operation is still running. Design handlers to be idempotent. ::: ::: warning When `PatchAllAsync` uses an uncached `ScriptPatch` or `PartialPatch` with a filter-only query (no explicit IDs), the `EntityChanged` message has `Id = null`. Cached/batch `PatchAllAsync` paths (`ActionPatch`, `JsonPatch`, or cached `ScriptPatch`/`PartialPatch`) send per-ID notifications for each modified document. Subscribers that depend on `msg.Id` to look up specific documents should handle the `null` case for uncached update-by-query, for example by re-querying the affected documents. ::: **In-process events:** The `DocumentsChanged` event fires for all patch types, but `args.Documents` is empty for `ScriptPatch`, `PartialPatch`, and single-doc `JsonPatch` because the modified document is not available client-side. Only single-document `ActionPatch` (`PatchAsync(id, ActionPatch)`) populates the documents list. Bulk operations — including `PatchAllAsync` and `PatchAsync(Ids)` for `ActionPatch`/`JsonPatch` (which delegates to `PatchAllAsync`) — fire `DocumentsChanged` with an **empty** documents list even though the documents may have been loaded during processing. The `DocumentsSaving` and `DocumentsSaved` events do **not** fire for patch operations. **Soft-delete detection:** Patch operations do not detect `IsDeleted` transitions. The `ChangeType` is always `Saved`, even if a patch sets `IsDeleted = true`. Use `SaveAsync` with `OriginalsEnabled = true` for soft-delete transition detection. ### Soft Delete Notification Logic When a document supports soft deletes (`ISupportSoftDeletes`), the notification system intelligently determines the `ChangeType`: ```csharp // If IsDeleted transitions from false to true: // ChangeType = Removed (not Saved) // This requires OriginalsEnabled = true to detect the transition public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { OriginalsEnabled = true; // Required for soft delete detection } } ``` **Example:** ```csharp var employee = await repository.GetByIdAsync(id); employee.IsDeleted = true; await repository.SaveAsync(employee); // EntityChanged message: // - Type: "Employee" // - Id: employee.Id // - ChangeType: Removed (not Saved!) ``` ## Subscribing to Notifications ### Basic Subscription ```csharp await messageBus.SubscribeAsync(async (msg, ct) => { Console.WriteLine($"{msg.Type} {msg.ChangeType}: {msg.Id}"); }); ``` ### Filter by Entity Type ```csharp await messageBus.SubscribeAsync(async (msg, ct) => { if (msg.Type == nameof(Employee)) { switch (msg.ChangeType) { case ChangeType.Added: await OnEmployeeAdded(msg.Id); break; case ChangeType.Saved: await OnEmployeeUpdated(msg.Id); break; case ChangeType.Removed: await OnEmployeeRemoved(msg.Id); break; } } }); ``` ### Real-Time UI Updates ```csharp public class EmployeeHub : Hub { private readonly IMessageBus _messageBus; public EmployeeHub(IMessageBus messageBus) { _messageBus = messageBus; } public override async Task OnConnectedAsync() { await _messageBus.SubscribeAsync(async (msg, ct) => { if (msg.Type == nameof(Employee)) { await Clients.All.SendAsync("EmployeeChanged", new { Id = msg.Id, ChangeType = msg.ChangeType.ToString() }); } }); await base.OnConnectedAsync(); } } ``` ## Repository Configuration ### NotificationsEnabled Enable or disable notifications at the repository level: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { NotificationsEnabled = true; // Default: true if message bus configured } } ``` ### NotificationDeliveryDelay Delay notification delivery to allow Elasticsearch indexing to complete: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { NotificationDeliveryDelay = TimeSpan.FromSeconds(2); } } ``` ::: warning Only set a delay if your message bus implementation supports delayed delivery. Message buses that don't support delayed delivery may silently drop messages. The in-memory message bus supports delayed delivery. ::: ### BatchNotifications Batch multiple notifications together: ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { BatchNotifications = true; } } ``` ### Per-Operation Control Disable notifications for specific operations: ```csharp // Disable notifications for bulk import await repository.AddAsync(employees, o => o.Notifications(false)); // Disable notifications for internal updates await repository.SaveAsync(employee, o => o.Notifications(false)); ``` ## BeforePublishEntityChanged Event Intercept and modify notifications before they're published. ### Event Arguments ```csharp public class BeforePublishEntityChangedEventArgs : CancelEventArgs { public EntityChanged Message { get; } public IReadOnlyRepository Repository { get; } // Inherited: bool Cancel { get; set; } } ``` ### Intercepting Notifications ```csharp public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { BeforePublishEntityChanged.AddHandler(OnBeforePublish); } private Task OnBeforePublish(object sender, BeforePublishEntityChangedEventArgs args) { // Add custom data to the notification args.Message.Data["TenantId"] = _tenantId; args.Message.Data["ModifiedBy"] = _currentUserId; return Task.CompletedTask; } } ``` ### Canceling Notifications ```csharp BeforePublishEntityChanged.AddHandler((sender, args) => { // Don't publish notifications for internal system changes if (args.Message.Data.ContainsKey("internal")) { args.Cancel = true; } // Don't publish for certain entity states // (Note: You'd need to fetch the document to check this) return Task.CompletedTask; }); ``` ### Adding Custom Data ```csharp BeforePublishEntityChanged.AddHandler((sender, args) => { // Add metadata to all notifications args.Message.Data["Timestamp"] = DateTime.UtcNow; args.Message.Data["Source"] = Environment.MachineName; return Task.CompletedTask; }); ``` ## In-Process Events vs Message Bus Foundatio.Repositories provides two notification mechanisms: ### In-Process Events Synchronous events fired within the same process: ```csharp repository.DocumentsChanged.AddHandler(async (sender, args) => { // Runs in the same process, same transaction context foreach (var doc in args.Documents) { await UpdateRelatedDataAsync(doc.Value); } }); ``` **Use for:** * Cache invalidation * Validation * Local side effects * Audit logging within the same service ### Message Bus Notifications Distributed messages published to `IMessageBus`: ```csharp await messageBus.SubscribeAsync(async (msg, ct) => { // Runs in any subscribed process await RefreshCacheAsync(msg.Id); }); ``` **Use for:** * Cross-service communication * Real-time UI updates * Event-driven architectures * Distributed cache invalidation ### Comparison | Aspect | In-Process Events | Message Bus | |--------|-------------------|-------------| | Scope | Same process | Distributed | | Timing | Synchronous | Asynchronous | | Reliability | Guaranteed | Depends on bus | | Use Case | Local side effects | Cross-service | | Access to Document | Full document for `SaveAsync` and single-doc `PatchAsync(id, ActionPatch)` only; empty list for all other patch paths | ID only (or `null` for type-level) | ## Distributed Cache Invalidation Use message bus notifications to invalidate cache across instances: ```csharp public class CacheInvalidationService : IHostedService { private readonly IMessageBus _messageBus; private readonly IEmployeeRepository _repository; public CacheInvalidationService(IMessageBus messageBus, IEmployeeRepository repository) { _messageBus = messageBus; _repository = repository; } public async Task StartAsync(CancellationToken cancellationToken) { await _messageBus.SubscribeAsync(async (msg, ct) => { if (msg.Type == nameof(Employee)) { // Invalidate local cache when other instances make changes await _repository.InvalidateCacheAsync(msg.Id); } }, cancellationToken); } public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; } ``` ## Event Sourcing Pattern Use notifications to build event-sourced systems: ```csharp public class EmployeeEventStore { private readonly IMessageBus _messageBus; private readonly IEventRepository _eventRepository; public EmployeeEventStore(IMessageBus messageBus, IEventRepository eventRepository) { _messageBus = messageBus; _eventRepository = eventRepository; } public async Task StartAsync() { await _messageBus.SubscribeAsync(async (msg, ct) => { if (msg.Type == nameof(Employee)) { await _eventRepository.AddAsync(new EmployeeEvent { EmployeeId = msg.Id, EventType = msg.ChangeType.ToString(), Timestamp = DateTime.UtcNow, Data = msg.Data }); } }); } } ``` ## Notification Flow Diagram ```mermaid sequenceDiagram participant App as Application participant Repo as Repository participant ES as Elasticsearch participant Bus as Message Bus participant Sub as Subscribers App->>Repo: SaveAsync(employee) Repo->>ES: Index document ES-->>Repo: Success Repo->>Repo: BeforePublishEntityChanged alt Not Cancelled Repo->>Bus: Publish EntityChanged Bus->>Sub: Deliver to subscribers Sub->>Sub: Process notification end Repo-->>App: Return result ``` ## Best Practices ### 1. Use OriginalsEnabled for Soft Delete Detection ```csharp OriginalsEnabled = true; // Detect IsDeleted transitions ``` ### 2. Add Custom Data for Context ```csharp BeforePublishEntityChanged.AddHandler((sender, args) => { args.Message.Data["UserId"] = _currentUserId; args.Message.Data["TenantId"] = _tenantId; return Task.CompletedTask; }); ``` ### 3. Handle Notification Failures Gracefully ```csharp await messageBus.SubscribeAsync(async (msg, ct) => { try { await ProcessNotificationAsync(msg); } catch (Exception ex) { _logger.LogError(ex, "Failed to process notification for {Type} {Id}", msg.Type, msg.Id); // Don't rethrow - allow other subscribers to process } }); ``` ### 4. Use NotificationDeliveryDelay for Eventual Consistency ```csharp NotificationDeliveryDelay = TimeSpan.FromSeconds(1); ``` ## Next Steps * [Caching](/guide/caching) - Distributed cache invalidation * [Configuration](/guide/configuration) - Notification configuration * [Repository Pattern](/guide/repository-pattern) - In-process events --- --- url: /guide/patch-operations.md --- # Patch Operations Foundatio.Repositories provides flexible patch operations for partial document updates without fetching the full document. This guide covers all patch types and their use cases. ## Overview Patch operations allow you to: * Update specific fields without loading the entire document * Execute atomic updates (counters, arrays) * Apply bulk updates efficiently * Reduce network traffic and conflicts ## Automatic Date Tracking For models implementing `IHaveDates`, all patch operations automatically set `UpdatedUtc` to the current time. This is consistent with how `AddAsync` and `SaveAsync` handle date tracking. * **`UpdatedUtc`** is set automatically on every patch operation — no manual intervention needed * **`CreatedUtc`** is not overwritten by patch operations under normal circumstances; it may be corrected if missing or invalid (for example, `DateTime.MinValue` or a future value), consistent with `SetDates` behavior * Works across all patch types: `PartialPatch`, `ScriptPatch`, `JsonPatch`, and `ActionPatch` ### Caller-provided UpdatedUtc For `ScriptPatch` and `PartialPatch`, if you explicitly provide `updatedUtc` in your script parameters or partial document, the repository respects your value and skips auto-injection. This is logged at `Debug` level. For `JsonPatch` and `ActionPatch`, `UpdatedUtc` is always set by the framework after applying your changes — matching `SaveAsync` semantics. If you need explicit control over the timestamp, use `ScriptPatch` or `PartialPatch`. ::: warning Stale timestamps on server-side retries For `ScriptPatch` and `PartialPatch`, the `UpdatedUtc` timestamp is captured once when the request is built and is fixed for any server-side retries (`RetryOnConflict`). If a version conflict causes Elasticsearch to retry the update, the retried write carries the **original timestamp**, not a fresh one. This means a document updated after several retries will have an `UpdatedUtc` that is slightly older than the actual commit time. For `JsonPatch` and `ActionPatch`, the timestamp is refreshed on each client-side retry since the entire operation (fetch + mutate + index) re-executes. This matches `SaveAsync` behavior, where `SetDates` is called once before the Elasticsearch request. In most scenarios the retry window is milliseconds, but callers relying on `UpdatedUtc` for strict ordering or audit trails should be aware of this. ::: ### Custom Date Fields If your model uses custom date fields instead of `IHaveDates` (e.g., a nested `MetaData.DateUpdatedUtc` property), you can opt into automatic date tracking by overriding three virtual hooks on your repository: ```csharp public class MyRepository : ElasticRepositoryBase { protected override bool HasDateTracking => true; protected override string GetUpdatedUtcFieldPath() { return InferField(d => ((IHaveDateMetaData)d).MetaData.DateUpdatedUtc); } protected override void SetDocumentDates(MyEntity document, TimeProvider timeProvider) { base.SetDocumentDates(document, timeProvider); if (document is IHaveDateMetaData metaDoc) { var utcNow = timeProvider.GetUtcNow().UtcDateTime; metaDoc.MetaData ??= new DateMetaData(); if (metaDoc.MetaData.DateCreatedUtc is null || metaDoc.MetaData.DateCreatedUtc == DateTime.MinValue || metaDoc.MetaData.DateCreatedUtc > utcNow) metaDoc.MetaData.DateCreatedUtc = utcNow; metaDoc.MetaData.DateUpdatedUtc = utcNow; } } } ``` | Hook | Purpose | Default Behavior | |------|---------|-----------------| | `HasDateTracking` | Gate for all date tracking logic | `true` when `T` implements `IHaveDates` | | `GetUpdatedUtcFieldPath()` | Returns the Elasticsearch field path for the updated timestamp | Returns the inferred `UpdatedUtc` field name. Throws `RepositoryException` if `HasDateTracking` is `true` but no field path is available | | `SetDocumentDates(T, TimeProvider)` | Sets date properties on the C# object (used by Add, Save, ActionPatch, JsonPatch bulk) | Sets `CreatedUtc` and `UpdatedUtc` on `IHaveDates` models | The `ApplyDateTracking` overloads for `ScriptPatch`, `PartialPatch`, and `JsonNode` are also virtual and can be overridden for full control over how dates are injected into each patch type. For nested fields, the script parameter key uses the last segment of the field path (e.g., `dateUpdatedUtc` for `metaData.dateUpdatedUtc`). ## Patch Types ### PartialPatch Update specific fields with new values: ```csharp // Update single field await repository.PatchAsync(id, new PartialPatch(new { Name = "John Smith" })); // Update multiple fields await repository.PatchAsync(id, new PartialPatch(new { Name = "John Smith", Age = 32, Department = "Engineering" })); // Update nested object await repository.PatchAsync(id, new PartialPatch(new { Address = new { City = "Seattle", State = "WA" } })); ``` ### ScriptPatch Use Elasticsearch Painless scripts for complex updates: ```csharp // Increment a counter await repository.PatchAsync(id, new ScriptPatch("ctx._source.viewCount++")); // Increment with parameter await repository.PatchAsync(id, new ScriptPatch("ctx._source.counter += params.amount") { Params = new Dictionary { ["amount"] = 5 } }); // Conditional update await repository.PatchAsync(id, new ScriptPatch(@" if (ctx._source.status == 'pending') { ctx._source.status = 'approved'; ctx._source.approvedAt = params.now; } ") { Params = new Dictionary { ["now"] = DateTime.UtcNow } }); // Array operations await repository.PatchAsync(id, new ScriptPatch("ctx._source.tags.add(params.tag)") { Params = new Dictionary { ["tag"] = "featured" } }); // Remove from array await repository.PatchAsync(id, new ScriptPatch("ctx._source.tags.remove(ctx._source.tags.indexOf(params.tag))") { Params = new Dictionary { ["tag"] = "draft" } }); ``` ### JsonPatch RFC 6902 JSON Patch operations: ```csharp using Foundatio.Repositories.JsonPatch; // Replace operation var patch = new PatchDocument( new ReplaceOperation { Path = "name", Value = "John Smith" } ); await repository.PatchAsync(id, new JsonPatch(patch)); // Multiple operations var patch = new PatchDocument( new ReplaceOperation { Path = "name", Value = "John Smith" }, new AddOperation { Path = "tags/-", Value = "senior" }, new RemoveOperation { Path = "tempField" } ); await repository.PatchAsync(id, new JsonPatch(patch)); ``` #### JSON Patch Operations | Operation | Description | Example | |-----------|-------------|---------| | `AddOperation` | Add value at path | `{ Path = "tags/-", Value = "new" }` | | `RemoveOperation` | Remove value at path | `{ Path = "tempField" }` | | `ReplaceOperation` | Replace value at path | `{ Path = "name", Value = "New Name" }` | | `MoveOperation` | Move value from one path to another | `{ From = "oldPath", Path = "newPath" }` | | `CopyOperation` | Copy value from one path to another | `{ From = "source", Path = "dest" }` | | `TestOperation` | Test value at path (fails if not equal) | `{ Path = "status", Value = "active" }` | ### ActionPatch Lambda-based patching for strongly-typed updates. Supports both `Action` (always writes) and `Func` (conditional write based on return value): ```csharp // Action — always treated as a modification await repository.PatchAsync(id, new ActionPatch(e => { e.Name = "John Smith"; e.Age = 32; })); // Func — return false to skip the write bool modified = await repository.PatchAsync(id, new ActionPatch(e => { if (e.Status == EmployeeStatus.Active) return false; e.Status = EmployeeStatus.Active; e.UpdatedBy = currentUserId; return true; })); ``` ::: tip `ActionPatch` fetches the document, applies the lambda, and saves it. For true partial updates without fetching, use `PartialPatch` or `ScriptPatch`. ::: ## Return Values All `PatchAsync` overloads return status information: * **`PatchAsync(Id, ...)`** returns `Task` -- `true` if the document was modified, `false` if the operation was treated as a no-op. * For `PartialPatch`, Elasticsearch's automatic noop detection reports `false` when the update does not change any field values (e.g., setting a field to its current value). However, models implementing `IHaveDates` have `UpdatedUtc` injected automatically on partial updates, so most `PartialPatch` calls will return `true` unless date tracking is disabled or the caller explicitly supplies an `UpdatedUtc` value that does not change. * For `ScriptPatch`, the operation is only a no-op when the script explicitly sets `ctx.op = 'none'`; simply reassigning the same value in a script is treated as a modification by Elasticsearch. The automatic date tracking script is appended after your script, but Elasticsearch evaluates `ctx.op` at the end of execution — so `ctx.op = 'none'` correctly prevents the write even with the appended timestamp assignment. * For `ActionPatch`, noop detection depends on the overload used. The `Func` overload respects the return value — `false` skips the Elasticsearch write entirely (no Index API call, no date tracking, no cache invalidation). The document is still fetched and the delegate executes, but the reindex step is avoided. The `Action` overload always assumes the document was modified. Empty actions (no callbacks) return `false`. * For `JsonPatch`, operations always return `true` (the get-modify-reindex pattern always writes). Empty operations (no patches) return `false`. * **`PatchAsync(Ids, ...)`** returns `Task` -- the number of documents actually modified (excludes no-ops as reported by the backend). * **`PatchAllAsync(...)`** returns `Task` — the number of documents modified by the query. Errors (document not found, version conflicts) throw exceptions rather than returning a status value. See [Error Handling](#error-handling) for details. ::: warning Noop Detection Limitations * **Date tracking**: Models implementing `IHaveDates` have `UpdatedUtc` set automatically on partial updates. This injected timestamp change means `PartialPatch` will almost always report `true` even when no other field values changed. * **Script patches**: Elasticsearch does not detect noops for script-based updates automatically. Your script must explicitly set `ctx.op = 'none'` to signal a no-op. See [Script Noop Example](#script-noop-example) below. * **JsonPatch**: Uses a get-modify-reindex pattern (Index API), so a write always occurs and `true` is always returned. * **ActionPatch (`Action`)**: The `Action` overload always assumes the document was modified. Use the `Func` overload and return `false` from your function to skip the reindex step for unchanged documents. ::: ### Script Noop Example Use `ctx.op = 'none'` in your Painless script to conditionally skip the update. Elasticsearch evaluates `ctx.op` after the entire script executes, so placing it anywhere in the script works — even when automatic date tracking appends a timestamp assignment. > **Note:** Painless uses `==` for equality comparison (Java-style), not `===` (JavaScript-style). The `===` operator is not valid in Painless. ```csharp // Only update if the value actually changes bool modified = await repository.PatchAsync(id, new ScriptPatch( """ if (ctx._source.status == params.newStatus) { ctx.op = 'none'; } else { ctx._source.status = params.newStatus; } """) { Params = new Dictionary { ["newStatus"] = "active" } }); // modified == false when the document already had status == "active" ``` For bulk operations, noop documents are excluded from the modified count: ```csharp long modifiedCount = await repository.PatchAsync( new Ids(id1, id2, id3), new ScriptPatch( """ if (ctx._source.status == params.newStatus) { ctx.op = 'none'; } else { ctx._source.status = params.newStatus; } """) { Params = new Dictionary { ["newStatus"] = "active" } }); // modifiedCount excludes documents that were already "active" ``` ### ActionPatch Noop Detection `ActionPatch` supports two modes — fire-and-forget with `Action`, or explicit noop signaling with `Func`. The `Func` overload lets your action return `false` to signal a no-op. When all actions return `false`, the write to Elasticsearch is skipped entirely — no Index API call, no date tracking update, no cache invalidation. The document is still fetched and the delegate executes, but the expensive reindex step is avoided. The `Action` overload always assumes the document was modified (returns `true` internally). Use `Func` when you want to conditionally skip writes: ```csharp // Action — always treated as modified (backward compatible) await repository.PatchAsync(id, new ActionPatch(e => e.Name = "Alice")); // Func — return false to skip the write when nothing changed bool modified = await repository.PatchAsync(id, new ActionPatch(e => { if (String.Equals(e.Name, "Alice", StringComparison.Ordinal)) return false; e.Name = "Alice"; return true; })); // modified == false when the employee's name was already "Alice" ``` For bulk operations (`PatchAsync(Ids, ...)` and `PatchAllAsync`), the same pattern applies — documents where all actions return `false` are skipped entirely and not sent to Elasticsearch. ## Patching Multiple Documents ### Patch by IDs ```csharp var ids = new[] { "emp-1", "emp-2", "emp-3" }; // Partial patch await repository.PatchAsync(ids, new PartialPatch(new { Department = "Engineering" })); // Script patch await repository.PatchAsync(ids, new ScriptPatch("ctx._source.reviewCount++")); ``` ### Patch by Query (PatchAllAsync) For `ISearchableRepository`: ```csharp // Update all matching documents long updated = await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Sales"), new PartialPatch(new { Region = "West" })); Console.WriteLine($"Updated {updated} documents"); // Increment counter for all matching await repository.PatchAllAsync( q => q.FieldEquals(e => e.Status, "active"), new ScriptPatch("ctx._source.loginCount++")); // Conditional bulk update await repository.PatchAllAsync( q => q.DateRange(null, DateTime.UtcNow.AddDays(-30), e => e.LastLoginUtc), new ScriptPatch(@" ctx._source.status = 'inactive'; ctx._source.deactivatedAt = params.now; ") { Params = new Dictionary { ["now"] = DateTime.UtcNow } }); ``` ## Patch Options ```csharp // Immediate consistency await repository.PatchAsync(id, patch, o => o.ImmediateConsistency()); // Skip version check await repository.PatchAsync(id, patch, o => o.SkipVersionCheck()); // Disable notifications await repository.PatchAsync(id, patch, o => o.Notifications(false)); ``` ## Common Patterns ### Counter Increment ```csharp // Atomic increment await repository.PatchAsync(id, new ScriptPatch("ctx._source.viewCount++")); // Increment by amount await repository.PatchAsync(id, new ScriptPatch("ctx._source.balance += params.amount") { Params = new Dictionary { ["amount"] = 100.00m } }); // Decrement with floor await repository.PatchAsync(id, new ScriptPatch(@" ctx._source.stock = Math.max(0, ctx._source.stock - params.quantity) ") { Params = new Dictionary { ["quantity"] = 5 } }); ``` ### Array Manipulation ```csharp // Add to array await repository.PatchAsync(id, new ScriptPatch("ctx._source.tags.add(params.tag)") { Params = new Dictionary { ["tag"] = "featured" } }); // Add if not exists await repository.PatchAsync(id, new ScriptPatch(@" if (!ctx._source.tags.contains(params.tag)) { ctx._source.tags.add(params.tag); } ") { Params = new Dictionary { ["tag"] = "featured" } }); // Remove from array await repository.PatchAsync(id, new ScriptPatch(@" if (ctx._source.tags.contains(params.tag)) { ctx._source.tags.remove(ctx._source.tags.indexOf(params.tag)); } ") { Params = new Dictionary { ["tag"] = "draft" } }); // Clear array await repository.PatchAsync(id, new ScriptPatch("ctx._source.tags.clear()")); ``` ### Conditional Updates ```csharp // Update only if condition is met await repository.PatchAsync(id, new ScriptPatch(@" if (ctx._source.status == params.expectedStatus) { ctx._source.status = params.newStatus; ctx._source.statusChangedAt = params.now; } ") { Params = new Dictionary { ["expectedStatus"] = "pending", ["newStatus"] = "approved", ["now"] = DateTime.UtcNow } }); // Increment only if below threshold await repository.PatchAsync(id, new ScriptPatch(@" if (ctx._source.failureCount < params.maxFailures) { ctx._source.failureCount++; } else { ctx._source.status = 'blocked'; } ") { Params = new Dictionary { ["maxFailures"] = 5 } }); ``` ### Timestamp Updates `UpdatedUtc` is set automatically for models implementing `IHaveDates` (see [Automatic Date Tracking](#automatic-date-tracking) above). For custom timestamp fields that are not part of `IHaveDates`, you can set them manually: ```csharp // Update a custom timestamp field await repository.PatchAsync(id, new ScriptPatch("ctx._source.lastAccessedAt = params.now") { Params = new Dictionary { ["now"] = DateTime.UtcNow } }); ``` ### Nested Object Updates ```csharp // Update nested field await repository.PatchAsync(id, new ScriptPatch("ctx._source.address.city = params.city") { Params = new Dictionary { ["city"] = "Seattle" } }); // Update entire nested object await repository.PatchAsync(id, new PartialPatch(new { Address = new { Street = "123 Main St", City = "Seattle", State = "WA", Zip = "98101" } })); ``` ## Cache Behavior ::: warning Important Patch operations invalidate cache by document ID, but **custom cache keys are NOT automatically invalidated**. ::: ```csharp // This invalidates cache by ID await repository.PatchAsync(id, patch); // But if you cached by email: var employee = await repository.FindOneAsync( q => q.FieldEquals(e => e.Email, email), o => o.Cache($"employee:email:{email}")); // And then patch the email: await repository.PatchAsync(id, new PartialPatch(new { Email = "new@example.com" })); // The cache key "employee:email:old@example.com" is NOT invalidated ``` **Solution:** Override `InvalidateCacheAsync` or manually invalidate: ```csharp await repository.PatchAsync(id, patch); await repository.InvalidateCacheAsync($"employee:email:{oldEmail}"); ``` ## Soft Delete Behavior Patch operations work on documents regardless of soft delete status: ```csharp // This will patch even if document is soft-deleted await repository.PatchAsync(id, patch); // PatchAllAsync respects SoftDeleteQueryMode for finding documents await repository.PatchAllAsync( q => q.FieldEquals(e => e.Status, "pending"), patch); // Only patches non-deleted documents by default // Include soft-deleted in bulk patch await repository.PatchAllAsync( q => q.FieldEquals(e => e.Status, "pending"), patch, o => o.IncludeSoftDeletes()); ``` ## Error Handling ```csharp try { await repository.PatchAsync(id, patch); } catch (DocumentNotFoundException ex) { // Document doesn't exist Console.WriteLine($"Document {ex.Id} not found"); } catch (VersionConflictDocumentException ex) { // Concurrent modification (if version checking enabled) Console.WriteLine($"Version conflict: {ex.Message}"); } ``` ### Exception Types by Patch Type All patch types can throw `DocumentNotFoundException` (HTTP 404) or `VersionConflictDocumentException` (HTTP 409). Other Elasticsearch errors produce a `DocumentException`. | Patch Type | 404 (Not Found) | 409 (Version Conflict) | Notes | |------------|-----------------|------------------------|-------| | `PartialPatch` | `DocumentNotFoundException` | `VersionConflictDocumentException` | Uses Elasticsearch Update API | | `ScriptPatch` | `DocumentNotFoundException` | `VersionConflictDocumentException` | Use `RetryOnConflict` for server-side retry | | `JsonPatch` | `DocumentNotFoundException` | `VersionConflictDocumentException` | Fetches document, applies patch, then indexes | | `ActionPatch` | `DocumentNotFoundException` | `VersionConflictDocumentException` | Fetches document, applies lambda, then indexes | ::: tip `ScriptPatch` and `PartialPatch` execute on the Elasticsearch node via the Update API, so you can set `RetryOnConflict` for server-side retries without a round trip. `JsonPatch` and `ActionPatch` perform a client-side fetch-mutate-index cycle, so version conflicts require the caller to retry the full operation. ::: ### Automatic Retry Behavior Transient errors (HTTP 429/503) are automatically retried with exponential backoff (up to 3 retries). Version conflicts (409) are **not** retried by the resilience policy — they indicate a genuine concurrency issue the caller should handle. ## Performance Considerations ### Use ScriptPatch for Atomic Operations ```csharp // Good: Atomic increment await repository.PatchAsync(id, new ScriptPatch("ctx._source.counter++")); // Less efficient: Fetch, modify, save var doc = await repository.GetByIdAsync(id); doc.Counter++; await repository.SaveAsync(doc); // Risk of lost updates ``` ### Batch Updates with PatchAllAsync ```csharp // Good: Single bulk operation await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Sales"), new PartialPatch(new { Region = "West" })); // Less efficient: Individual patches var employees = await repository.FindAsync(q => q.FieldEquals(e => e.Department, "Sales")); foreach (var emp in employees.Documents) { await repository.PatchAsync(emp.Id, new PartialPatch(new { Region = "West" })); } ``` ### Cache Invalidation Patch operations handle cache invalidation differently depending on whether the full document is available: **Document-based invalidation** (supports custom `InvalidateCacheAsync` overrides): * `ActionPatch` — single-doc and bulk: the modified `T` is passed to `InvalidateCacheAsync(IEnumerable)` and `OnDocumentsChangedAsync`, enabling custom cache key strategies based on document properties. * `JsonPatch` — bulk (`PatchAllAsync`): deserialized documents are tracked and used for document-based invalidation after successful indexing. **ID-based invalidation** (removes cache entries by document ID only): * `ScriptPatch` — all paths: executes server-side on Elasticsearch; the modified document is never returned to the client. * `PartialPatch` — all paths: same as `ScriptPatch`; the Update API does not return the full document. * `JsonPatch` — single-doc (`PatchAsync(Id)`): uses the low-level client with raw `JToken`, so a typed `T` is not available. ::: warning ID-Based Invalidation Limitation If your repository overrides `InvalidateCacheAsync(IReadOnlyCollection>)` to compute custom cache keys from document properties (e.g., composite keys, secondary lookups), those overrides will **not** fire for `ScriptPatch`, `PartialPatch`, or single-doc `JsonPatch`. Only the standard ID-based cache key is removed. Consider using `ActionPatch` with `Func` if you need full document-based cache invalidation for conditional updates. ::: ## Notifications Patch operations publish `EntityChanged` messages (with `ChangeType.Saved`) to the message bus and fire the in-process `DocumentsChanged` event when documents are modified. The behavior varies by patch type and method. ### PatchAsync (Single Document) All patch types publish an `EntityChanged` message with the document's ID when the patch modifies the document: ```csharp await repository.PatchAsync(id, new ScriptPatch("ctx._source.name = 'Changed';")); // EntityChanged { Type = "Employee", Id = "", ChangeType = Saved } ``` No-op patches (e.g., `ScriptPatch` with `ctx.op = 'none'`, `ActionPatch` returning `false`, or empty operations) do **not** send notifications. ::: warning DocumentsChanged fires with an empty document list For `ScriptPatch`, `PartialPatch`, and single-doc `JsonPatch`, the `DocumentsChanged` event fires but `args.Documents` is **empty** because the modified document is not available client-side. Only single-document `ActionPatch` (`PatchAsync(id, ActionPatch)`) provides the full document in `DocumentsChanged`. Bulk operations — including `PatchAllAsync` and `PatchAsync(Ids)` for `ActionPatch`/`JsonPatch` (which delegates to `PatchAllAsync`) — also fire `DocumentsChanged` with an **empty** documents list. If your `DocumentsChanged` handler iterates over `args.Documents`, it will see zero items for all patch types except single-document `ActionPatch`. ::: ### PatchAsync (Multi-ID) `PatchAsync(Ids, ...)` behavior depends on the patch type: * **`ScriptPatch` / `PartialPatch`**: Sends one `EntityChanged` message **per modified ID**. IDs that result in a noop (e.g., `ScriptPatch` with `ctx.op = 'none'`) are excluded from notifications: ```csharp await repository.PatchAsync(new Ids(id1, id2, id3), new ScriptPatch("ctx._source.name = 'Changed';")); // EntityChanged { Type = "Employee", Id = "", ChangeType = Saved } // EntityChanged { Type = "Employee", Id = "", ChangeType = Saved } // EntityChanged { Type = "Employee", Id = "", ChangeType = Saved } ``` * **`JsonPatch` / `ActionPatch`**: The multi-ID overload delegates to `PatchAllAsync` internally with the IDs set in the query. This means query-based notification rules apply — one `EntityChanged` per ID in the query when at least one document is modified, regardless of whether a particular ID was a noop. ### PatchAllAsync (Query-Based) `PatchAllAsync` notification behavior depends on whether the query contains explicit IDs and whether caching is enabled: | Query Type | Caching Enabled | Notification Behavior | |------------|----------------|----------------------| | Any query | Yes | One `EntityChanged` message **per modified ID** (sent per-batch as documents are processed) | | Query with explicit IDs | No | One `EntityChanged` message **per ID** | | Filter-only query (no IDs) | No | One `EntityChanged` message with `Id = null` (type-level notification) | ::: info Per-batch delivery Notifications are sent incrementally per batch as documents are processed, not after the entire operation completes. Subscribers may see `EntityChanged` messages arriving while the operation is still processing later batches. Design subscribers to be idempotent. ::: ```csharp // Uncached update-by-query with filter-only query: sends a single type-level notification (Id = null) await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Sales"), new PartialPatch(new { Region = "West" })); // EntityChanged { Type = "Employee", Id = null, ChangeType = Saved } // Cached/batch path with filter-only query: sends per-ID notifications for modified documents await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Sales"), new ActionPatch(e => { e.Region = "West"; })); // EntityChanged { Type = "Employee", Id = "", ChangeType = Saved } // EntityChanged { Type = "Employee", Id = "", ChangeType = Saved } // ... ``` ::: warning Type-level notifications apply only to uncached update-by-query When `PatchAllAsync` uses an uncached `ScriptPatch` or `PartialPatch` with a filter-only query (no explicit IDs), subscribers receive a single `EntityChanged` message with `Id = null`. Cached/batch `PatchAllAsync` paths (`ActionPatch`, `JsonPatch`, or cached `ScriptPatch`/`PartialPatch`) send per-ID notifications for each modified document. If your subscriber needs specific document IDs, either use a cached patch type or restructure the query to include explicit IDs. ::: ### DocumentsSaving / DocumentsSaved Events Unlike `SaveAsync`, patch operations do **not** fire `DocumentsSaving` or `DocumentsSaved` events. Only `DocumentsChanged` is fired. This is important if you have handlers on those events expecting to see patched documents. ### Suppressing Notifications ```csharp await repository.PatchAsync(id, patch, o => o.Notifications(false)); await repository.PatchAllAsync(q => q, patch, o => o.Notifications(false)); ``` See [Message Bus](/guide/message-bus) for full details on `EntityChanged` subscriptions, `BeforePublishEntityChanged`, and notification configuration. ### Soft Delete Detection on Patches Patch operations do **not** detect soft-delete transitions (`IsDeleted: false → true`). The `ChangeType` for a patch is always `Saved`, even if the patch sets `IsDeleted = true`. Soft-delete transition detection requires `OriginalsEnabled` and only works with `SaveAsync`. ## Next Steps * [CRUD Operations](/guide/crud-operations) - Full document operations * [Caching](/guide/caching) - Cache behavior with patches * [Soft Deletes](/guide/soft-deletes) - Patch behavior with soft deletes * [Message Bus](/guide/message-bus) - Notification subscriptions and configuration --- --- url: /guide/soft-deletes.md --- # Soft Deletes Foundatio.Repositories provides built-in soft delete support, allowing you to mark documents as deleted without physically removing them. This guide covers the soft delete behavior across all repository APIs. ## Overview Soft deletes allow you to: * Mark documents as deleted without permanent removal * Restore deleted documents * Query deleted documents when needed * Maintain audit trails and data recovery options ## Enabling Soft Deletes Implement `ISupportSoftDeletes` on your entity: ```csharp using Foundatio.Repositories.Models; public class Employee : IIdentity, IHaveDates, ISupportSoftDeletes { public string Id { get; set; } = string.Empty; public string Name { get; set; } = string.Empty; public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } public bool IsDeleted { get; set; } // Required by ISupportSoftDeletes } ``` The repository automatically: * Detects `ISupportSoftDeletes` implementation * Adds `IsDeleted` to the index mapping * Filters queries based on `SoftDeleteQueryMode` ## SoftDeleteQueryMode Control how soft-deleted documents are handled in queries: ```csharp public enum SoftDeleteQueryMode { ActiveOnly, // Only IsDeleted = false (default) DeletedOnly, // Only IsDeleted = true All // All documents regardless of IsDeleted } ``` ### Setting the Mode ```csharp // Include soft-deleted documents var results = await repository.FindAsync(query, o => o.IncludeSoftDeletes()); // Only deleted documents var results = await repository.FindAsync(query, o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); // Explicitly active only (default) var results = await repository.FindAsync(query, o => o.SoftDeleteMode(SoftDeleteQueryMode.ActiveOnly)); ``` ## Soft Delete vs Hard Delete ### Soft Delete Mark a document as deleted (recoverable): ```csharp var employee = await repository.GetByIdAsync(id); employee.IsDeleted = true; await repository.SaveAsync(employee); ``` ### Hard Delete Permanently remove a document (not recoverable): ```csharp await repository.RemoveAsync(id); // or await repository.RemoveAsync(employee); ``` ## API Behavior Reference ### Read Operations | API | Default Behavior | Respects SoftDeleteMode | Notes | |-----|------------------|------------------------|-------| | `GetByIdAsync` | Filters deleted | Yes | Returns `null` for soft-deleted | | `GetByIdsAsync` | Filters deleted | Yes | Excludes soft-deleted from results | | `GetAllAsync` | Filters deleted | Yes | Delegates to `FindAsync` | | `FindAsync` | Filters deleted | Yes | Uses `SoftDeletesQueryBuilder` | | `FindOneAsync` | Filters deleted | Yes | Uses `SoftDeletesQueryBuilder` | | `CountAsync` | Filters deleted | Yes | Only counts active documents | | `ExistsAsync` | Filters deleted | Yes | Uses search for soft-delete models | ### Write Operations | API | Behavior | Notes | |-----|----------|-------| | `AddAsync` | N/A | New documents typically have `IsDeleted = false` | | `SaveAsync` | Use for soft delete | Set `IsDeleted = true` and save | | `RemoveAsync` | **HARD DELETE** | Permanently removes document | | `RemoveAllAsync` | **HARD DELETE** | Permanently removes matching documents | | `PatchAsync` | No filtering | Operates directly by ID | | `PatchAllAsync` | Filters deleted | Query respects `SoftDeleteMode` | ## Detailed API Behavior ### GetByIdAsync ```csharp // Default: Returns null for soft-deleted documents var employee = await repository.GetByIdAsync(id); // Include soft-deleted var employee = await repository.GetByIdAsync(id, o => o.IncludeSoftDeletes()); // Only if deleted var employee = await repository.GetByIdAsync(id, o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); ``` ### GetByIdsAsync ```csharp var ids = new[] { "emp-1", "emp-2", "emp-3" }; // Default: Excludes soft-deleted var employees = await repository.GetByIdsAsync(ids); // Include all var employees = await repository.GetByIdsAsync(ids, o => o.IncludeSoftDeletes()); ``` ### FindAsync ```csharp // Default: Only active documents var results = await repository.FindAsync(q => q.FieldEquals(e => e.Department, "Engineering")); // Include soft-deleted var results = await repository.FindAsync( q => q.FieldEquals(e => e.Department, "Engineering"), o => o.IncludeSoftDeletes()); // Only deleted var results = await repository.FindAsync( q => q.FieldEquals(e => e.Department, "Engineering"), o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); ``` ### CountAsync ```csharp // Count active only long activeCount = await repository.CountAsync(); // Count all including deleted var result = await repository.CountAsync(q => q, o => o.IncludeSoftDeletes()); long totalCount = result.Total; // Count deleted only var deletedResult = await repository.CountAsync(q => q, o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); long deletedCount = deletedResult.Total; ``` ### ExistsAsync ```csharp // Check if active document exists bool exists = await repository.ExistsAsync(id); // Check if document exists (including deleted) bool exists = await repository.ExistsAsync(id, o => o.IncludeSoftDeletes()); ``` ### RemoveAsync (Hard Delete) ::: warning `RemoveAsync` performs a **hard delete** - the document is permanently removed from Elasticsearch. ::: ```csharp // Permanently delete await repository.RemoveAsync(id); await repository.RemoveAsync(employee); await repository.RemoveAsync(employees); ``` ### RemoveAllAsync (Hard Delete) ```csharp // Permanently delete all matching (respects SoftDeleteMode for finding) long deleted = await repository.RemoveAllAsync( q => q.FieldEquals(e => e.Status, "inactive")); // Delete including soft-deleted long deleted = await repository.RemoveAllAsync( q => q.FieldEquals(e => e.Status, "inactive"), o => o.IncludeSoftDeletes()); ``` ### PatchAsync Patch operations work on documents regardless of soft delete status: ```csharp // This will patch even if document is soft-deleted await repository.PatchAsync(id, new PartialPatch(new { Name = "Updated" })); ``` ### PatchAllAsync Query respects `SoftDeleteMode`: ```csharp // Only patches active documents await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Sales"), new PartialPatch(new { Region = "West" })); // Patch including soft-deleted await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Sales"), new PartialPatch(new { Region = "West" }), o => o.IncludeSoftDeletes()); ``` ## Soft Delete Operations ### Soft Delete a Document ```csharp var employee = await repository.GetByIdAsync(id); employee.IsDeleted = true; await repository.SaveAsync(employee); ``` ### Restore a Soft-Deleted Document ```csharp // Get the deleted document var employee = await repository.GetByIdAsync(id, o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); // Restore it employee.IsDeleted = false; await repository.SaveAsync(employee); ``` ### Bulk Soft Delete ```csharp // Using PatchAllAsync await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Closed"), new PartialPatch(new { IsDeleted = true })); // Or using BatchProcessAsync await repository.BatchProcessAsync( q => q.FieldEquals(e => e.Department, "Closed"), async batch => { foreach (var emp in batch.Documents) { emp.IsDeleted = true; } await repository.SaveAsync(batch.Documents); return true; }); ``` ### Bulk Restore ```csharp await repository.PatchAllAsync( q => q.FieldEquals(e => e.Department, "Reopened"), new PartialPatch(new { IsDeleted = false }), o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); ``` ## Parent-Child Soft Delete Filtering When using parent-child relationships, children are automatically filtered when their parent is soft-deleted: ```csharp // Parent is soft-deleted parent.IsDeleted = true; await parentRepository.SaveAsync(parent); // Children are now filtered out (even though they're not deleted) var children = await childRepository.FindAsync(q => q.ParentId("parent-child", parent.Id)); // Returns empty - children are filtered because parent is deleted // Restore parent parent.IsDeleted = false; await parentRepository.SaveAsync(parent); // Children are now visible again var children = await childRepository.FindAsync(q => q.ParentId("parent-child", parent.Id)); // Returns children ``` ## EntityChanged Notifications When a document is soft-deleted, the notification system sends `ChangeType.Removed`: ```csharp // Enable originals tracking (required for soft delete detection) public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { OriginalsEnabled = true; } } // When soft-deleting: employee.IsDeleted = true; await repository.SaveAsync(employee); // EntityChanged message: // - ChangeType: Removed (not Saved!) // - Id: employee.Id // - Type: "Employee" ``` ::: tip Set `OriginalsEnabled = true` in your repository to enable soft delete transition detection. Without this, soft deletes will send `ChangeType.Saved` instead of `ChangeType.Removed`. ::: ## Cache Behavior The repository maintains a special cache list to handle eventual consistency: ### How It Works 1. When a document is soft-deleted, its ID is added to a `"deleted"` list in cache 2. The list has a 30-second TTL 3. Queries automatically exclude IDs in the `"deleted"` list ```csharp // When you soft-delete: employee.IsDeleted = true; await repository.SaveAsync(employee); // Internally: // 1. Document is indexed to Elasticsearch // 2. ID is added to "deleted" cache list (30s TTL) // 3. Subsequent queries exclude this ID from results ``` ### Purpose This handles the eventual consistency window where: 1. Document is soft-deleted 2. Elasticsearch hasn't indexed the change yet 3. Cache knows about the deletion 4. Queries correctly exclude the document After 30 seconds, Elasticsearch should have indexed the change, and the cache entry expires. ## Query Filtering Implementation The `SoftDeletesQueryBuilder` automatically adds filters to queries: ```csharp // For ActiveOnly mode: // Adds: { "term": { "isDeleted": false } } // For DeletedOnly mode: // Adds: { "term": { "isDeleted": true } } // For All mode: // No filter added ``` ## Common Patterns ### Audit Trail ```csharp public class Employee : IIdentity, IHaveDates, ISupportSoftDeletes { public string Id { get; set; } public bool IsDeleted { get; set; } public DateTime? DeletedUtc { get; set; } public string DeletedBy { get; set; } // ... other properties } // When soft-deleting: employee.IsDeleted = true; employee.DeletedUtc = DateTime.UtcNow; employee.DeletedBy = currentUserId; await repository.SaveAsync(employee); ``` ### Scheduled Hard Delete ```csharp // Delete documents that have been soft-deleted for more than 30 days var cutoffDate = DateTime.UtcNow.AddDays(-30); await repository.RemoveAllAsync( q => q .FieldEquals(e => e.IsDeleted, true) .DateRange(null, cutoffDate, e => e.DeletedUtc), o => o.IncludeSoftDeletes()); ``` ### Recycle Bin UI ```csharp // Get deleted items for recycle bin var deletedItems = await repository.FindAsync( q => q.SortExpression("-deletedUtc"), o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly).PageLimit(50)); // Restore selected items foreach (var id in selectedIds) { await repository.PatchAsync(id, new PartialPatch(new { IsDeleted = false })); } // Permanently delete selected items foreach (var id in selectedIds) { var item = await repository.GetByIdAsync(id, o => o.SoftDeleteMode(SoftDeleteQueryMode.DeletedOnly)); if (item != null) await repository.RemoveAsync(item); } ``` ## Summary Table | Operation | Soft-Deleted Documents | Notes | |-----------|----------------------|-------| | `GetByIdAsync` | Filtered by default | Use `IncludeSoftDeletes()` to include | | `GetByIdsAsync` | Filtered by default | Use `IncludeSoftDeletes()` to include | | `FindAsync` | Filtered by default | Use `SoftDeleteMode()` to control | | `CountAsync` | Filtered by default | Use `SoftDeleteMode()` to control | | `ExistsAsync` | Filtered by default | Use `IncludeSoftDeletes()` to include | | `SaveAsync` | Use to soft delete | Set `IsDeleted = true` | | `RemoveAsync` | **Hard deletes** | Permanently removes | | `RemoveAllAsync` | **Hard deletes** | Query respects mode | | `PatchAsync` | No filtering | Works on any document | | `PatchAllAsync` | Query filtered | Use `IncludeSoftDeletes()` | ## Next Steps * [Message Bus](/guide/message-bus) - Soft delete notifications * [Caching](/guide/caching) - Soft delete cache behavior * [Configuration](/guide/configuration) - Soft delete configuration --- --- url: /guide/versioning.md --- # Versioning Foundatio.Repositories provides optimistic concurrency control through document versioning. This guide covers how versioning works and how to handle conflicts. ## Overview Versioning prevents lost updates when multiple processes modify the same document simultaneously. When enabled, the repository checks that the document version matches before saving. ## Enabling Versioning Implement `IVersioned` on your entity: ```csharp using Foundatio.Repositories.Models; public class Employee : IIdentity, IHaveDates, IVersioned { public string Id { get; set; } = string.Empty; public string Name { get; set; } = string.Empty; public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } public string Version { get; set; } = string.Empty; // Required by IVersioned } ``` ## How Versioning Works ### Version Format The version is stored as `"primaryTerm:sequenceNumber"` using Elasticsearch's sequence numbers: ```csharp // Example version: "1:42" // - primaryTerm: 1 // - sequenceNumber: 42 ``` ### Read-Modify-Write Cycle ```csharp // 1. Read document (version is populated) var employee = await repository.GetByIdAsync(id); Console.WriteLine($"Version: {employee.Version}"); // e.g., "1:42" // 2. Modify document employee.Name = "John Smith"; // 3. Save document (version is checked) await repository.SaveAsync(employee); // If version matches: Save succeeds, version updated to "1:43" // If version doesn't match: VersionConflictDocumentException thrown ``` ### Conflict Detection ```mermaid sequenceDiagram participant A as Process A participant B as Process B participant ES as Elasticsearch A->>ES: GetByIdAsync (version: 1:42) B->>ES: GetByIdAsync (version: 1:42) A->>A: Modify document B->>B: Modify document A->>ES: SaveAsync (expects 1:42) ES-->>A: Success (new version: 1:43) B->>ES: SaveAsync (expects 1:42) ES-->>B: VersionConflictException (current: 1:43) ``` ## Handling Version Conflicts ### Basic Error Handling ```csharp try { await repository.SaveAsync(employee); } catch (VersionConflictDocumentException ex) { Console.WriteLine($"Version conflict: {ex.Message}"); // Handle conflict... } ``` ### Retry Pattern ```csharp public async Task UpdateEmployeeWithRetry(string id, Action update, int maxRetries = 3) { int retries = maxRetries; while (retries > 0) { try { var employee = await repository.GetByIdAsync(id); if (employee == null) throw new DocumentNotFoundException(id); update(employee); await repository.SaveAsync(employee); return; // Success } catch (VersionConflictDocumentException) { retries--; if (retries == 0) throw; // Wait before retry (exponential backoff) await Task.Delay(TimeSpan.FromMilliseconds(100 * (maxRetries - retries))); } } } // Usage await UpdateEmployeeWithRetry(id, emp => emp.Name = "John Smith"); ``` ### Merge Strategy ```csharp public async Task UpdateEmployeeWithMerge(string id, Action update) { while (true) { try { var employee = await repository.GetByIdAsync(id); if (employee == null) throw new DocumentNotFoundException(id); update(employee); return await repository.SaveAsync(employee); } catch (VersionConflictDocumentException) { // Refresh and retry - the update action will be applied to fresh data continue; } } } ``` ### Last-Write-Wins Skip version checking when you want the last write to win: ```csharp await repository.SaveAsync(employee, o => o.SkipVersionCheck()); ``` ## Skipping Version Check ### Per-Operation ```csharp // Skip version check for this operation await repository.SaveAsync(employee, o => o.SkipVersionCheck()); // Explicitly control version checking await repository.SaveAsync(employee, o => o.VersionCheck(false)); ``` ### Use Cases for Skipping * **Bulk imports** - Trusted data from verified sources * **System updates** - Internal updates that should always succeed * **Migrations** - Data transformations during schema changes * **Last-write-wins scenarios** - When conflicts don't matter ::: warning Only skip version checking when you're certain that lost updates are acceptable. ::: ## Version in Patch Operations Patch operations also support version checking: ```csharp // Patch with version check (default) await repository.PatchAsync(id, patch); // Patch without version check await repository.PatchAsync(id, patch, o => o.SkipVersionCheck()); ``` ## Atomic Operations For truly atomic operations, use `ScriptPatch`: ```csharp // Atomic increment - no version conflict possible await repository.PatchAsync(id, new ScriptPatch("ctx._source.counter++")); // Atomic conditional update await repository.PatchAsync(id, new ScriptPatch(@" if (ctx._source.status == 'pending') { ctx._source.status = 'approved'; } ")); ``` ## Version in Find Results When querying, versions are available in the hits: ```csharp var results = await repository.FindAsync(query); foreach (var hit in results.Hits) { Console.WriteLine($"ID: {hit.Id}, Version: {hit.Version}"); Console.WriteLine($"Document Version: {hit.Document.Version}"); } ``` ## Common Patterns ### Optimistic Locking Service ```csharp public class OptimisticLockingService where T : class, IIdentity, IVersioned, new() { private readonly IRepository _repository; private readonly int _maxRetries; public OptimisticLockingService(IRepository repository, int maxRetries = 3) { _repository = repository; _maxRetries = maxRetries; } public async Task UpdateAsync(string id, Func updateAction) { int retries = _maxRetries; while (true) { var document = await _repository.GetByIdAsync(id); if (document == null) throw new DocumentNotFoundException(id); await updateAction(document); try { return await _repository.SaveAsync(document); } catch (VersionConflictDocumentException) { retries--; if (retries == 0) throw; await Task.Delay(TimeSpan.FromMilliseconds(50 * (_maxRetries - retries))); } } } } // Usage var service = new OptimisticLockingService(repository); var updated = await service.UpdateAsync(id, async emp => { emp.Name = "John Smith"; emp.UpdatedBy = currentUserId; }); ``` ### Compare-and-Swap ```csharp public async Task CompareAndSwapAsync(string id, string expectedValue, string newValue) { try { var employee = await repository.GetByIdAsync(id); if (employee == null || employee.Status != expectedValue) return false; employee.Status = newValue; await repository.SaveAsync(employee); return true; } catch (VersionConflictDocumentException) { return false; } } // Usage bool success = await CompareAndSwapAsync(id, "pending", "approved"); ``` ### Conditional Update ```csharp public async Task<(bool Success, T Document)> ConditionalUpdateAsync( string id, Func condition, Action update) where T : class, IIdentity, IVersioned, new() { var document = await repository.GetByIdAsync(id); if (document == null) return (false, null); if (!condition(document)) return (false, document); update(document); try { var saved = await repository.SaveAsync(document); return (true, saved); } catch (VersionConflictDocumentException) { return (false, document); } } // Usage var (success, employee) = await ConditionalUpdateAsync( id, emp => emp.Status == "pending", emp => emp.Status = "approved"); ``` ## Bulk Operations and Partial Failures When saving multiple versioned documents in a single call, some may succeed while others hit version conflicts. The repository handles this as a **partial failure**: 1. **Successful documents** are fully processed — versions updated, events fired, cache populated. 2. **Conflicting documents** have their cache entries left unchanged — the concurrent writer that caused the conflict handles its own cache update. 3. A `VersionConflictDocumentException` is thrown after processing all successes. ```csharp try { await repository.SaveAsync(employees); } catch (VersionConflictDocumentException ex) { // Successful documents were saved and notified. // Conflicting documents need to be re-fetched and retried. _logger.LogWarning(ex, "Some documents had version conflicts"); } ``` ::: warning Version conflicts (HTTP 409) on `AddAsync`/`SaveAsync` are **not** automatically retried by the resilience policy. Transient errors (HTTP 429/503) are retried with exponential backoff, but conflict resolution is the caller's responsibility. ::: ## Best Practices ### 1. Always Handle Version Conflicts ```csharp try { await repository.SaveAsync(employee); } catch (VersionConflictDocumentException) { _logger.LogWarning("Version conflict for {Id}, retrying...", employee.Id); // Implement retry logic } ``` ### 2. Use Atomic Operations When Possible ```csharp // Prefer atomic script over read-modify-write await repository.PatchAsync(id, new ScriptPatch("ctx._source.counter++")); ``` ### 3. Keep Retry Counts Reasonable ```csharp // 3-5 retries is usually sufficient const int MaxRetries = 3; ``` ### 4. Use Exponential Backoff ```csharp await Task.Delay(TimeSpan.FromMilliseconds(100 * Math.Pow(2, attempt))); ``` ### 5. Log Conflicts for Monitoring ```csharp catch (VersionConflictDocumentException) { _metrics.IncrementCounter("version_conflicts", new { entity = typeof(T).Name }); _logger.LogWarning("Version conflict for {Type}", typeof(T).Name); } ``` ## Next Steps * [CRUD Operations](/guide/crud-operations) - Save operations with versioning * [Patch Operations](/guide/patch-operations) - Atomic updates * [Configuration](/guide/configuration) - Version check configuration --- --- url: /guide/index-management.md --- # Index Management Foundatio.Repositories provides flexible index management strategies for different use cases. This guide covers index types, configuration, and maintenance. ## Index Types ### Index\ Basic index for simple entities: ```csharp public sealed class EmployeeIndex : Index { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees") { } public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.CompanyId) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) ); } } ``` ### VersionedIndex\ Index with schema versioning for evolving schemas: ```csharp public sealed class EmployeeIndex : VersionedIndex { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) { } public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.CompanyId) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) .Keyword(e => e.Department) // Added in v2 ); } } ``` **Index naming:** * Version 1: `employees-v1` * Version 2: `employees-v2` * Alias: `employees` (points to current version) ### DailyIndex\ Time-series index with daily partitioning: ```csharp public sealed class LogEventIndex : DailyIndex { public LogEventIndex(IElasticConfiguration configuration) : base(configuration, "logs", version: 1) { MaxIndexAge = TimeSpan.FromDays(90); DiscardExpiredIndexes = true; } public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.Level) .Text(e => e.Message) ); } } ``` **Index naming:** * `logs-v1-2024.01.15` * `logs-v1-2024.01.16` * Alias: `logs` (points to all indexes) ### MonthlyIndex\ Time-series index with monthly partitioning: ```csharp public sealed class AuditLogIndex : MonthlyIndex { public AuditLogIndex(IElasticConfiguration configuration) : base(configuration, "audit", version: 1) { MaxIndexAge = TimeSpan.FromDays(365); DiscardExpiredIndexes = true; } } ``` **Index naming:** * `audit-v1-2024.01` * `audit-v1-2024.02` ## How Time-Series Indexes Work `DailyIndex` and `MonthlyIndex` spread documents across many small time-partitioned indexes rather than one large index. Understanding how a document's index is **picked at write time** and **resolved at read time** explains the whole model — including why there is normally exactly **one** index per time period and no parallel copies of the same data. ### One index per period, not parallel copies A common question is whether the library keeps multiple copies of an index in parallel or processes one index at a time with cleanup. The answer is the latter: * **Steady state:** exactly **one** physical index exists per time period (per day for `DailyIndex`, per month for `MonthlyIndex`). The umbrella alias unions all of them so the repository can query them as if they were a single index. * **Retention:** as periods age past `MaxIndexAge`, their indexes are removed from the aliases and then deleted (see [Retention Policy](#retention-policy-for-time-series-indexes)). Old data is cleaned up one index at a time, not held indefinitely. * **The only time two copies of the same period coexist** is transiently during a [version reindex](#version-upgrade-process) (e.g. `logs-v1-2024.01.15` → `logs-v2-2024.01.15`). After the reindex succeeds, the old version is discarded when `DiscardIndexesOnReindex` is `true` (the default). ### Three naming layers Time-series indexes use three distinct name layers. Knowing which is which is the key to understanding routing: | Layer | Example | Points to | Used for | |---|---|---|---| | **Physical index** | `logs-v1-2024.01.15` | Actual Lucene index on disk (version encoded) | Where documents physically live | | **Dated alias** | `logs-2024.01.15` | The current version's physical index for that one day | Routing a single document's read/write | | **Umbrella alias** | `logs` | All current, non-expired physical indexes | Querying across all periods | | **Windowed alias** | `logs-last-7-days` | Physical indexes within a rolling window | Fixed-window queries (see [Time-Based Aliases](#time-based-aliases)) | Because read/write routing targets the **dated alias** (unversioned), the physical version can change underneath (via reindex) without changing how the repository addresses documents. ### Picking the index at write time When you write a document (`AddAsync`, `SaveAsync`, bulk operations), the library derives the target index from the document's **date**, resolved in this order (`DailyIndex.GetIndex` / `_getDocumentDateUtc`): 1. If the document id is an [ObjectId](/guide/crud-operations), its embedded **creation timestamp** is used. `CreateDocumentId` generates an ObjectId that encodes the document date, so the id and its index stay consistent. 2. Otherwise, if the model implements `IHaveCreatedDate`, its `CreatedUtc` value is used. 3. You can override resolution entirely by passing a `getDocumentDateUtc` delegate to the index constructor. That date maps to a dated alias (`logs-2024.01.15` for daily, `logs-2024.01` for monthly). Before the write, `EnsureIndexAsync` creates the physical index for that period **if it does not already exist** and attaches its aliases in the same call: * the **dated alias** (`logs-2024.01.15`), * the **umbrella alias** (`logs`), and * any **windowed aliases** whose age window still includes that date. Writes are grouped by resolved index, so a bulk insert spanning several days fans out into one write per dated index. ```mermaid flowchart TD Doc["Document to write"] --> Date["Resolve document date\nObjectId.CreationTime → CreatedUtc → custom func"] Date --> Dated["Target = dated alias\nlogs-2024.01.15"] Dated --> Age{"Date older than\nMaxIndexAge?"} Age -->|Yes| Reject["Throw: Index max age exceeded"] Age -->|No| Exists{"Physical index\nlogs-v1-2024.01.15\nexists?"} Exists -->|No| Create["Create physical index +\nattach umbrella / dated / windowed aliases"] Exists -->|Yes| Write Create --> Write["Index document into that single dated index"] ``` ::: warning Writing to an already-expired period fails If a document's date is older than `MaxIndexAge`, `EnsureDateIndexAsync` throws `ArgumentException: Index max age exceeded` rather than silently recreating a period that retention has already reclaimed. See [Preventing Writes to Expired Indexes](#preventing-writes-to-expired-indexes). ::: ### Resolving the index at read time Reads resolve differently depending on whether you look up a single document or run a query: * **Single-document lookups** (`GetByIdAsync`, `ExistsAsync`, and id-based `PatchAsync` / `RemoveAsync`) route directly to **one** dated alias by parsing the ObjectId in the id back into its creation date. This avoids scanning every period. If the document is not found there and the index has multiple partitions, the repository falls back to a query across the umbrella alias. * **Queries** (`FindAsync`, `CountAsync`, `PatchAllAsync`, `RemoveAllAsync`) resolve their target indexes via `GetIndexesByQuery`: * `.Index("name")` targets explicit index/alias names. * `.Index(start, end)` expands to the list of dated aliases in that range (partition pruning). * When neither is set — or the range is too wide (see [Large Range Fallback](#large-range-fallback)) — the query targets the **umbrella alias** covering all periods. ```mermaid flowchart TD subgraph Single["Single-document lookup (GetByIdAsync)"] Id["id (ObjectId)"] --> Parse["Parse creation date"] Parse --> Route["Route to dated alias\nlogs-2024.01.15"] Route --> Found{"Found?"} Found -->|Yes| Return["Return document"] Found -->|"No + multiple indexes"| Umbrella1["Fallback: query umbrella alias logs"] end subgraph Query["Query (FindAsync / CountAsync)"] Q["Query"] --> HasRange{".Index(start, end) set\nand range within threshold?"} HasRange -->|Yes| Prune["Target only matching dated aliases"] HasRange -->|"No / too wide"| Umbrella2["Target umbrella alias logs (all periods)"] end ``` ### Alias management and retention over time `MaintainIndexesAsync` (run on a schedule via [`MaintainIndexesJob`](/guide/jobs)) keeps aliases in sync with `MaxIndexAge`: * Current-version, non-expired indexes are **added** to the umbrella and any matching windowed aliases. * Expired indexes (age past `MaxIndexAge`) and superseded versions are **removed** from all aliases so queries stop hitting them. * When `DiscardExpiredIndexes` is `true`, expired physical indexes are then **deleted**. ```mermaid flowchart LR Maintain["MaintainIndexesAsync()"] --> Update["UpdateAliasesAsync:\nadd current/non-expired,\nremove expired + old versions"] Maintain --> Age{"DiscardExpiredIndexes\nand age > MaxIndexAge?"} Age -->|Yes| Delete["Delete expired physical index"] Age -->|No| Keep["Keep index, only drop from aliases"] ``` See [Retention Policy for Time-Series Indexes](#retention-policy-for-time-series-indexes) for configuration details. ## Querying Time-Series Indexes ### Index Selection vs. Document Filtering When working with `DailyIndex` or `MonthlyIndex`, two separate mechanisms control what data is returned: * **`.Index(start, end)`** — selects which physical index partitions to query. Without this, all partitions are queried via the umbrella alias. * **`.DateRange(start, end, field)`** — filters documents within the targeted indexes by a date field value. These must be set independently. `DateRange` alone does not narrow index selection. ```csharp var start = DateTime.UtcNow.AddDays(-7); var end = DateTime.UtcNow; var results = await repository.FindAsync(q => q .Index(start, end) // target only the relevant partitions .DateRange(start, end, e => e.CreatedUtc) // filter documents within those partitions ); ``` Omitting `.Index()` is correct but less efficient — the query runs against all partitions and relies solely on the `DateRange` filter to narrow results. ### Large Range Fallback Generating an individual index name for each day or month in a very wide range would produce an excessively long list. To avoid this, `.Index(start, end)` falls back to the umbrella alias (which covers all partitions) when the range is too broad: | Index type | Threshold | Behavior | |---|---|---| | `DailyIndex` | Range >= 3 months, or exceeds `MaxIndexAge` | Falls back to alias (all partitions) | | `MonthlyIndex` | Range > 1 year, or exceeds `MaxIndexAge` | Falls back to alias (all partitions) | In the fallback case, Elasticsearch receives the alias name rather than a list of specific index names. The query is still executed correctly, and the `.DateRange()` filter still restricts the returned documents — there is just no partition pruning at the index-routing level. ```csharp // This range is 4 months — exceeds the DailyIndex threshold of 3 months. // GetIndexes returns an empty list, so the query targets the "logs" alias instead. var results = await repository.FindAsync(q => q .Index(DateTime.UtcNow.AddMonths(-4), DateTime.UtcNow) .DateRange(DateTime.UtcNow.AddMonths(-4), DateTime.UtcNow, e => e.CreatedUtc) ); ``` ### Best Practices for Time-Series Queries 1. **Always pair `.Index()` with `.DateRange()`** — `.Index()` prunes partitions, `.DateRange()` filters documents. Both are needed for correct and efficient queries. 2. **Keep ranges within the fallback threshold** — queries within 3 months (daily) or 1 year (monthly) benefit from partition pruning. Wider ranges fall back to alias-level querying. 3. **Use time-based aliases for fixed windows** — for recurring queries like "last 7 days" or "last 30 days", configure named aliases via `AddAlias()` to avoid computing index ranges at query time. ```csharp // In index configuration public LogEventIndex(IElasticConfiguration configuration) : base(configuration, "logs", version: 1) { MaxIndexAge = TimeSpan.FromDays(90); AddAlias("logs-last-7-days", TimeSpan.FromDays(7)); AddAlias("logs-last-30-days", TimeSpan.FromDays(30)); } // In queries — use the alias directly instead of computing a range var results = await repository.FindAsync(q => q.Index("logs-last-7-days")); ``` ## Index Configuration ### Index Settings ```csharp public override void ConfigureIndex(CreateIndexRequestDescriptor idx) { base.ConfigureIndex(idx.Settings(s => s .NumberOfShards(3) .NumberOfReplicas(1) .RefreshInterval(new Duration(TimeSpan.FromSeconds(5))) .Analysis(a => a .AddSortNormalizer() ))); } ``` ### Index Mapping ```csharp public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) // Disable dynamic mapping .Properties(p => p .SetupDefaults() // Configure Id, CreatedUtc, UpdatedUtc, IsDeleted // Keyword fields (exact match, aggregations) .Keyword(e => e.CompanyId) .Keyword(e => e.Status) // Text fields with keywords (full-text + exact match) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) .Text(e => e.Email, t => t.AddKeywordAndSortFields()) // Numeric fields .IntegerNumber(e => e.Age) .DoubleNumber(e => e.Salary) // Date fields .Date(e => e.HireDate) // Boolean fields .Boolean(e => e.IsActive) // Nested objects .Nested(e => e.Addresses, n => n .Properties(ap => ap .Keyword(a => a.City) .Keyword(a => a.Country) )) ); } ``` ### SetupDefaults Extension The `SetupDefaults()` extension configures common fields: ```csharp .Properties(p => p.SetupDefaults()) ``` This configures: * `Id` as keyword * `CreatedUtc` as date * `UpdatedUtc` as date * `IsDeleted` as boolean (if `ISupportSoftDeletes`) * `Version` as keyword (if `IVersioned`) ## Schema Versioning ### How Versioned Indexes Work When you use `VersionedIndex`, the library manages schema evolution through a versioning system: 1. **Index Naming**: Each version creates a separate index (e.g., `employees-v1`, `employees-v2`) 2. **Alias Management**: An alias (`employees`) always points to the current version 3. **Reindexing**: When you increment the version, data is migrated by a reindex — but that reindex has to be explicitly triggered (see [What actually triggers a reindex](#what-actually-triggers-a-reindex)); nothing runs it for you automatically ```mermaid graph LR A[Application] -->|queries| B[employees alias] B -->|points to| C[employees-v2] D[employees-v1] -->|reindex| C style D fill:#f9f,stroke:#333,stroke-dasharray: 5 5 ``` ::: tip When to bump the version Only increment the version when you need to **change an existing field's mapping type** (e.g., `text` to `keyword`) or run a **data transformation** via reindex script. Elasticsearch [does not allow in-place type changes](https://www.elastic.co/docs/manage-data/data-store/mapping/update-mappings-examples) on existing fields. **Adding a mapping for a brand-new field does NOT require a version bump.** See [Mapping Lifecycle](#mapping-lifecycle) for the full breakdown of how mappings are applied per index type, including important differences for `DailyIndex`/`MonthlyIndex`. ::: ### Version Upgrade Process ::: info Single index vs. time-series The steps below describe a single-index type (`Index` / `VersionedIndex`), where one physical index is swapped. For `DailyIndex` / `MonthlyIndex`, the same steps run **per dated partition, one at a time** — see [Version Upgrades for Time-Series Indexes](#version-upgrades-for-time-series-indexes-daily-monthly) for exactly when each old partition is dropped. ::: When an index's version is incremented, the actual upgrade is always these 5 steps. The only variable is *what triggers them* — see [What actually triggers a reindex](#what-actually-triggers-a-reindex) below, because it is **not** simply "calling `ConfigureIndexesAsync()`": 1. **New Index Creation**: Creates `employees-v2` with the new mapping 2. **Reindex Task**: Elasticsearch's reindex API copies data from v1 to v2 3. **Script Execution**: Any reindex scripts transform data during migration 4. **Alias Switch**: The `employees` alias is atomically switched from v1 to v2 5. **Old Index Cleanup**: If `DiscardIndexesOnReindex` is true, v1 is deleted ```csharp // Step 1: Increment version and add migration scripts public sealed class EmployeeIndex : VersionedIndex { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) // Changed from 1 to 2 { // Scripts run during reindex from v1 to v2 RenameFieldScript(2, "dept", "department"); RemoveFieldScript(2, "legacyField"); } } // Step 2: Run the reindex directly — deterministic and awaitable. // ConfigureIndexesAsync()'s default only enqueues a work item; see below. await configuration.ReindexAsync(); ``` ::: warning `ConfigureIndexesAsync()` does not reindex inline `ConfigureIndexesAsync()` (default `beginReindexingOutdated: true`) does **not** run the 5 steps above itself — it only *enqueues* a `ReindexWorkItem`. Something else (a queue worker with `ReindexWorkItemHandler` registered) has to dequeue and actually run it. If you never configured an `IQueue` on `ElasticConfiguration`, this throws `InvalidOperationException: Must specify work item queue and lock provider in order to migrate index versions.` the moment it finds an outdated index. See [What actually triggers a reindex](#what-actually-triggers-a-reindex) for the recommended, direct alternative. ::: ### Version Upgrades for Time-Series Indexes (Daily/Monthly) `DailyIndex` and `MonthlyIndex` store one physical index per time period, so bumping the version has to migrate **every** existing partition. It does this **one partition at a time**, and each partition's old index is deleted as the *final step of that partition's own reindex* — before the next partition begins. It never creates new copies of all partitions first and then bulk-deletes the originals. ::: tip One at a time, not all-at-once Peak extra disk usage during a time-series version upgrade is roughly **one partition** (the one currently being migrated), not a full duplicate of the entire dataset. Already-migrated partitions have their old index deleted; not-yet-migrated partitions still have only their original. ::: Trigger a time-series version upgrade explicitly with `configuration.ReindexAsync()` (or `auditIndex.ReindexAsync()`). It runs inline (awaitable) and reports progress through the optional callback: ```csharp // After bumping the index version (e.g. new MonthlyIndex(configuration, version: 2)): await configuration.ReindexAsync((progress, message) => { logger.LogInformation("Reindex {Progress:F0}%: {Message}", progress, message); return Task.CompletedTask; }); // Or reindex a single time-series index directly: await auditIndex.ReindexAsync(); ``` ::: warning Trigger time-series upgrades with `ReindexAsync()`, not `ConfigureIndexesAsync()` Unlike a single `VersionedIndex`, a `DailyIndex` / `MonthlyIndex` is **not** migrated by the automatic background reindex that `ConfigureIndexesAsync(beginReindexingOutdated: true)` enqueues — that path targets the unversioned base name (`audit-v1`), which does not match the dated partitions (`audit-v1-2024.01`, …). Always trigger a time-series version upgrade explicitly with `configuration.ReindexAsync()` or `index.ReindexAsync()`, which runs the per-partition loop described below. ::: `ReindexAsync` then: 1. **Acquires a distributed lock** keyed on the alias (`reindex:audit`) so only one reindex runs at a time. The lock is auto-renewed on every progress callback. 2. **Lists all v1 partitions** and orders them **oldest → newest** by index date. 3. **For each partition**, runs the full sequence to completion before moving to the next: 1. Create `audit-v2-2024.01` with the new mapping. 2. Reindex documents from `audit-v1-2024.01` into it (first pass). 3. **Swap aliases** — atomically remove `audit-v1-2024.01` from every alias and add `audit-v2-2024.01`. Reads for that month now hit v2. 4. **Second-pass catch-up** copies any documents written during the first pass (see [Second-Pass Catch-Up Strategy](#second-pass-catch-up-strategy)). 5. **Delete `audit-v1-2024.01`** (conditional — see below). 4. Move on to `audit-v1-2024.02`, then `audit-v1-2024.03`, and so on. Partitions already past `MaxIndexAge` are **skipped** (left for [retention/maintenance](#retention-policy-for-time-series-indexes) to clean up rather than reindexed). During the migration the umbrella alias (`audit`) transparently spans both already-migrated (v2) and not-yet-migrated (v1) partitions, so reads and writes keep working the entire time. ```mermaid flowchart TD Start["Bump version → configuration.ReindexAsync()"] --> Lock["Acquire distributed lock (keyed on alias)"] Lock --> List["List v1 partitions,\nordered oldest → newest"] List --> Loop{"More partitions?"} Loop -->|No| Done["Upgrade complete"] Loop -->|Yes| Expired{"Partition past\nMaxIndexAge?"} Expired -->|Yes| Loop Expired -->|No| Create["Create audit-v2-YYYY.MM"] Create --> Reindex["Reindex v1 → v2 (first pass)"] Reindex --> Swap["Swap aliases:\nremove v1 partition, add v2 partition"] Swap --> Catchup["Second-pass catch-up"] Catchup --> Check{"DiscardIndexesOnReindex\nAND no failures\nAND new count ≥ old count?"} Check -->|Yes| Delete["Delete audit-v1-YYYY.MM"] Check -->|No| Keep["Keep old partition\n(inspect / retry)"] Delete --> Loop Keep --> Loop ``` #### When the old partition is deleted The old index for a period is deleted at the very end of *that period's* reindex (~98–99% progress), and **only** when all of the following hold: * `DiscardIndexesOnReindex` is `true` (the default). * Neither the first nor the second reindex pass reported any failures. * The new partition's document count is **greater than or equal to** the old partition's count (a safety check against data loss). If any condition fails, the old partition is **retained** so you can inspect or retry it, and the alias already points at the new partition. Because deletion happens per-partition immediately after that partition's data is verified, the originals are never all held simultaneously and then dropped in one batch. #### What actually triggers a reindex No mechanism in the library starts a reindex automatically — there is no background timer, hosted service, or auto-discovered job. A version bump only takes effect once something explicitly calls it. There are three ways to do that: 1. **Call `configuration.ReindexAsync()` / `index.ReindexAsync()` directly.** This is the deterministic, inline, awaitable path described throughout this section — one partition at a time — and it's what every reindex test in this repo uses. Run it from a deploy step, an admin endpoint, a one-off console command, or a job you write and schedule yourself. **This is the recommended way to run a version upgrade**, time-series or not. 2. **The `beginReindexingOutdated: true` default on `ConfigureIndexesAsync()`.** This does **not** perform a reindex itself — it only *enqueues* a `ReindexWorkItem` (see [Configure Indexes](#configure-indexes)). For that work item to actually run, two more things must be true: (a) a real `IQueue` was passed into `ElasticConfiguration`'s constructor, and (b) something in the app is dequeuing work items with `ReindexWorkItemHandler` registered to handle `ReindexWorkItem`s. **Neither is wired up by the library.** If no queue is configured and an index turns out to be outdated, `ConfigureIndexesAsync()` throws `InvalidOperationException: Must specify work item queue and lock provider in order to migrate index versions.` — which is why this repo's own [sample app](https://github.com/FoundatioFx/Foundatio.Repositories/blob/main/samples/Foundatio.SampleApp/Server/Repositories/Configuration/ElasticExtensions.cs) calls `ConfigureIndexesAsync(beginReindexingOutdated: false)` instead of relying on the default. Even fully wired up, this path is a **no-op for time-series indexes** (see the warning above) — the enqueued work item names the non-dated base index, which matches no dated partition. 3. **`ElasticMigrationJobBase`** (`Jobs/ElasticMigrationJob.cs`) is an abstract helper class for a repeatable "run migrations, then reindex everything outdated" job — it correctly calls `ConfigureIndexesAsync(beginReindexingOutdated: false)` (sidestepping the no-op queue path) and then `index.ReindexAsync()` for every outdated index. **It is opt-in scaffolding, not something registered or run automatically.** Nothing in the library subclasses it, schedules it, or references it, and no consuming application in this repository — including its own sample app — derives from it. Derive from it and register it with your own job runner for a repeatable/scheduled job; for a one-time upgrade, calling `ReindexAsync()` directly (option 1) is simpler and is what's actually tested. For a manual, one-time upgrade — such as bumping the version on a monthly audit index — call `configuration.ReindexAsync()` or `auditIndex.ReindexAsync()` explicitly when ready to run it. `ConfigureIndexesAsync()`'s default does not perform the upgrade, and no built-in job runs it automatically. Neither of the following reindexes time-series data: `MaintainIndexesJob` (aliases/retention only), and the `ReindexWorkItemHandler` queue path for daily/monthly indexes (the enqueued work item's name doesn't match any dated partition). #### Concurrency: within an index, one partition at a time **Within a single index** a reindex is **strictly sequential** — one partition at a time, with no parallel fan-out: | Level | Behavior | Where | |---|---|---| | **Partitions within an index** | `ReindexAsync` iterates partitions in a single `await`ed `foreach`; the next partition never starts until the current one finishes (including its delete). | `DailyIndex.ReindexAsync` | | **The Elasticsearch reindex itself** | Each partition is copied with a **single, unsliced** `_reindex` task. The library does not set `slices`, so there is no parallel sub-task fan-out; it submits the task and polls until it completes. | `ElasticReindexer.InternalReindexAsync` | **Across different indexes** it depends on how you trigger it: `configuration.ReindexAsync()` processes indexes **sequentially** (one index fully finishes before the next starts), while `ElasticMigrationJob` reindexes them **in parallel** (`Task.WhenAll`, one task per outdated index). Either way each index is internally sequential, and a **distributed lock keyed on the alias** (`reindex:audit`) guarantees a given index is never reindexed by two runners at once — even across multiple application instances (pods, workers). The lock is held for 20 minutes and auto-renewed on every progress callback, so long partition copies keep it alive. ::: tip Predictable, bounded disk usage per index Within one index the upgrade only ever duplicates **one partition at a time**, so bumping a single index (e.g. `audit`) needs roughly one extra partition of headroom regardless of how many partitions it has. If several indexes reindex in parallel (via `ElasticMigrationJob`), peak extra disk is about the sum of one in-flight partition per concurrently-migrating index. Wall-clock time scales with partition count; run during off-peak hours if needed. ::: #### Multiple versions and interrupted upgrades In normal operation only **two** versions of a period ever coexist, and only transiently — the old partition and the new one — during that single period's reindex. The process is designed to be **resumable and idempotent**: * The **lowest version still present** is treated as the current version (`GetCurrentVersionAsync`), and each run processes only the partitions still on that version. Partitions that were already migrated are excluded automatically, so re-running never redoes completed work. * If a run is interrupted — a process restart, a failure on one partition, a lost lock — just **run it again**. It picks up the remaining old partitions and continues, oldest first. A partition whose reindex failed keeps its old index (the delete is gated on success), so nothing is lost. * Reindex scripts **compose across skipped versions**: going straight from v1 to v3 applies the v2 and v3 scripts in order, so transformations are never skipped. * If partitions end up at genuinely mixed versions (for example a v1→v2 upgrade was interrupted and you have since bumped to v3), each run advances the oldest cohort one step; run the reindex until `GetCurrentVersionAsync()` equals the target `Version`. The migration job converges this over repeated runs. Throughout, the umbrella alias spans whatever the current partitions are, so reads and writes keep working even while the index is a mix of versions. #### Recovering from a rolling restart mid-upgrade A reindex can be interrupted at any point — a deploy recycles the pod running it, a node is drained, the process crashes. Re-running `configuration.ReindexAsync()` (or `index.ReindexAsync()`) afterward recovers cleanly, without manual cleanup, for the following reasons: * **The lock expires; nobody has to release it.** The distributed lock (`reindex:audit`) is held for 20 minutes and renewed on every progress callback. If the process holding it dies, the lock is never explicitly released — it simply expires 20 minutes after the last renewal. A new instance's call to `ReindexAsync()` waits for the lock (up to 30 minutes) and then proceeds. * **The Elasticsearch-side copy isn't tied to the calling process.** Each partition's copy runs as an asynchronous Elasticsearch task (`wait_for_completion=false`); the library only polls it for progress. That task lives in the cluster's task manager, so if the .NET process dies while polling, the copy already running in Elasticsearch is unaffected and keeps going independently. * **A retried first pass copies only the delta.** On retry, the first pass queries the new partition for the most recent document it already contains and reindexes only source documents at or after that point, rather than recopying the whole period. If the new partition is empty (nothing had landed before the interruption), the retry does a full copy, same as an initial run. * **A partition whose alias was already swapped is still found and finished.** Partitions to migrate are discovered by matching physical index names, not by current alias membership. If the process died after the alias swap but before the old partition's delete, the next run still finds that now-orphaned old partition, reruns its (now-cheap) resume copy and alias swap, and deletes it — reaching the same end state as an uninterrupted run. * **Two instances never migrate the same index at once.** The alias-keyed lock caps a given index to one active reindex cluster-wide. If a rolling restart briefly leaves two instances both calling `ReindexAsync()` for the same index, one holds the lock while the other waits; once the first finishes, the current version has already advanced, so the second call's version check finds nothing left to do and returns immediately. #### When do writes flip to the new partition — and is there a gap? Writes for a period target the **unversioned dated alias** (e.g. `audit-2024.01`), so they flip when that alias is repointed: 1. During the **first pass**, the dated alias still points to the old partition, so any concurrent writes for that period land in **v1**. 2. When the first pass finishes (~91–92%), **every alias pointing at the old partition — the dated alias, the umbrella alias, and any windowed aliases — is repointed to the new partition in a single `UpdateAliases` call**. From that instant, new writes for that period land in **v2**. 3. The **second-pass catch-up** then copies anything written to v1 during the first pass into v2. **Is there a gap?** * **No aliasing gap.** The remove-old and add-new actions are submitted together in one `UpdateAliases` request, which Elasticsearch applies **atomically**. The alias is never pointing at zero indexes (or at both), so reads and writes always resolve to exactly one partition — there is no window where a write fails to route or a read sees nothing. * **No lost-write gap for append-only data.** Documents written to the old partition during the first pass are picked up by the second-pass catch-up, which runs *after* the swap and copies every document with a timestamp (or ObjectId creation time) at or after a start time captured ~1 second before the reindex began. After the swap the old partition receives no new writes, and `Conflicts=proceed` keeps the catch-up from failing on documents already copied. This is why a `TimestampField` or ObjectId-format IDs are recommended (see [Second-Pass Catch-Up Strategy](#second-pass-catch-up-strategy)) — they let the catch-up find late writes precisely. Only the currently-reindexing period has this brief hand-off; periods not yet reached still write to v1, and periods already migrated write to v2 — all through the same unchanging dated-alias names. #### Why partitions are processed oldest → newest Partitions are always migrated in ascending date order (`GetIndexesAsync` sorts by `DateUtc`). This is deliberate, and it matters most for exactly the append-only time-series workloads these indexes are built for (audit logs, events): * **Least write contention and near-empty catch-up.** In a time-series workload new documents land in the **current** period; older periods are effectively immutable (and writing to a period past `MaxIndexAge` throws). Migrating the old, static partitions first means their first pass captures everything and the [second-pass catch-up](#second-pass-catch-up-strategy) has little or nothing to copy. The one volatile partition — today/this month — is migrated **last**, so the short window where concurrent writes must be caught up is isolated at the very end instead of being reopened repeatedly. * **Progressive, predictable disk reclamation.** Since each old partition is deleted before the next starts, disk is freed starting with your oldest data and continues steadily — helpful when the whole reason for going one-at-a-time is limited headroom. * **Deterministic and resumable.** The "current version" is the **lowest** version still present, and each run lists only the partitions still on that old version — already-migrated partitions are excluded automatically. So if a run is interrupted or retried, it simply resumes with the remaining old partitions in the same order, without redoing completed work. (This deterministic ordering was introduced as an index-management stability fix and has been the behavior since.) ### Field Operations During Reindex `RenameFieldScript` and `RemoveFieldScript` generate Painless scripts that automatically handle field names with special characters. Standard identifiers use dot-notation (e.g. `ctx._source.data.field`), while field names containing hyphens, `@`, spaces, or other non-identifier characters automatically use bracket notation (e.g. `ctx._source['@timestamp']`). Field paths cannot contain single quotes (`'`), backslashes (`\`), or control characters (such as newlines), since these would break Painless string literals. #### Rename a Field Use `RenameFieldScript` to rename a field during reindex: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) { // Rename 'dept' to 'department' in version 2 RenameFieldScript(2, "dept", "department"); // By default, the original field is removed // To keep both fields: RenameFieldScript(2, "oldName", "newName", removeOriginal: false); } ``` The generated Painless script: ```javascript if (ctx._source.containsKey('dept')) { ctx._source.department = ctx._source.dept; } if (ctx._source.containsKey('dept')) { ctx._source.remove('dept'); } ``` #### Rename a Nested Field `RenameFieldScript` supports dotted paths for nested properties: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) { RenameFieldScript(2, "data.oldField", "data.newField"); // Deeply nested paths are also supported: RenameFieldScript(2, "metadata.author.name", "metadata.author.displayName"); } ``` The generated Painless script for nested paths includes null-safety guards: ```javascript if (ctx._source.data != null && ctx._source.data.containsKey('oldField')) { if (ctx._source.data == null) { ctx._source.data = [:]; } ctx._source.data.newField = ctx._source.data.oldField; } if (ctx._source.data != null && ctx._source.data.containsKey('oldField')) { ctx._source.data.remove('oldField'); } ``` #### Remove a Field Use `RemoveFieldScript` to remove a field: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 3) { RemoveFieldScript(3, "deprecatedField"); } ``` #### Remove a Nested Field `RemoveFieldScript` also supports dotted paths: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 3) { RemoveFieldScript(3, "data.legacyField"); } ``` #### Custom Transformation Use `AddReindexScript` for complex transformations: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 4) { // Custom Painless script for complex transformation AddReindexScript(4, @" // Combine first and last name if (ctx._source.containsKey('firstName') && ctx._source.containsKey('lastName')) { ctx._source.fullName = ctx._source.firstName + ' ' + ctx._source.lastName; } // Convert status string to boolean if (ctx._source.containsKey('status')) { ctx._source.isActive = ctx._source.status == 'active'; ctx._source.remove('status'); } // Set default values if (!ctx._source.containsKey('createdUtc')) { ctx._source.createdUtc = '2024-01-01T00:00:00Z'; } "); } ``` ### Multi-Version Migration Scripts are applied incrementally. When upgrading, only scripts with a version greater than the current index version (and less than or equal to the target version) are applied. If upgrading from v1 to v3, both v2 and v3 scripts run: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 3) { // v2 scripts (run when upgrading from v1) RenameFieldScript(2, "dept", "department"); // v3 scripts (run when upgrading from v1 or v2) AddReindexScript(3, "ctx._source.version = 3;"); } ``` When a single script applies, it is sent directly to Elasticsearch. When multiple scripts apply, they are each wrapped in a named function and called sequentially: ```javascript void f000(def ctx) { /* v2 rename script */ } void f001(def ctx) { /* v2 remove script */ } void f002(def ctx) { /* v3 custom script */ } f000(ctx); f001(ctx); f002(ctx); ``` Note that `RenameFieldScript` with `removeOriginal: true` (the default) generates **two** scripts at the same version number — one to copy the value and one to remove the original field. Both are included in the combined script. #### Skipping Over Multiple Versions If an index is multiple versions behind (e.g., v1 upgrading to v5), all intermediate scripts run in order: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 5) { RenameFieldScript(2, "dept", "department"); // v2 RemoveFieldScript(3, "data.legacyField"); // v3 RenameFieldScript(4, "data.oldField", "data.newField"); // v4 AddReindexScript(5, "ctx._source.migrated = true;"); // v5 } ``` When upgrading from v1 to v5, scripts for v2 through v5 all apply. When upgrading from v3 to v5, only v4 and v5 scripts apply. Scripts for versions at or below the current version are always skipped. #### Moving Fields Between Objects You can rename fields across different parent objects: ```csharp RenameFieldScript(2, "data.oldField", "meta.newField"); // Move between parents RenameFieldScript(3, "data.name", "displayName"); // Promote nested to top-level RenameFieldScript(4, "companyName", "data.company"); // Demote top-level to nested ``` ### Controlling Old Index Deletion ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) { // Delete old index after successful reindex (default: true) DiscardIndexesOnReindex = true; // Keep old index for rollback capability // DiscardIndexesOnReindex = false; } ``` Even with `DiscardIndexesOnReindex = true`, the old index is only deleted when the reindex reported **no failures** and the new index's document count is **greater than or equal to** the old index's count. If either check fails, the old index is kept so you can inspect or retry. For time-series indexes this evaluation happens independently per dated partition — see [When the old partition is deleted](#when-the-old-partition-is-deleted). ### Reindex Progress Monitoring Monitor reindex progress with a callback: ```csharp await configuration.ReindexAsync(async (progress, message) => { _logger.LogInformation("Reindex {Progress}%: {Message}", progress, message); // Update UI or metrics await UpdateProgressAsync(progress, message); }); ``` ### Throttling Reindex Load Internally, reindexing uses Elasticsearch's `_reindex` API, which reads and writes documents in bulk batches (default 1000 documents per batch, unlimited throughput). For indexes with large documents, the default batch size can produce bulk sub-requests large enough to exceed a node's [indexing pressure](https://www.elastic.co/docs/reference/elasticsearch/configuration-reference/indexing-pressure-settings) memory limit (10% of heap by default), causing Elasticsearch to reject the request with an `es_rejected_execution_exception` (`rejected execution of coordinating operation`). See [Troubleshooting: Reindex Rejected Due to Indexing Pressure](./troubleshooting.md#reindex-rejected-due-to-indexing-pressure) for how to recognize this error. Set `ReindexBatchSize` and/or `ReindexRequestsPerSecond` on the index to reduce the size and rate of these internal batches: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) { // Read/write at most 200 documents per internal bulk batch (default: 1000) ReindexBatchSize = 200; // Throttle to ~500 documents/second (default: unlimited) ReindexRequestsPerSecond = 500; } ``` Both properties are `null` by default, which preserves the current Elasticsearch defaults. They apply to `Index`, `VersionedIndex`, `DailyIndex`, and `MonthlyIndex` since all of them build on the same reindex work item. Lower `ReindexBatchSize` first if you're seeing indexing pressure rejections; add `ReindexRequestsPerSecond` on top of that if the cluster is still under load from other traffic during the reindex. Both values must be greater than zero when set - `ReindexAsync` throws `ArgumentOutOfRangeException` immediately for a zero, negative, infinite, or `NaN` value rather than sending an invalid request to Elasticsearch. A low `ReindexRequestsPerSecond` makes Elasticsearch pause longer between internal batches (roughly `ReindexBatchSize` ÷ `ReindexRequestsPerSecond`) to honor the throttle. Reindex progress is monitored by polling for status, and a reindex that reports no progress for too long is treated as stalled and abandoned - the threshold defaults to 10 minutes but automatically extends (with a 3x safety margin) when a configured throttle would otherwise make that inter-batch pause exceed it, so a slow but healthy, intentionally throttled reindex isn't cancelled by mistake. ### Error Handling During Reindex Failed documents are stored in an error index (`employees-v2-error`): ```csharp // Query failed documents var errorIndex = "employees-v2-error"; var failures = await _client.SearchAsync(s => s.Index(errorIndex)); foreach (var failure in failures.Documents) { // Handle failed document _logger.LogError("Failed to reindex: {Document}", failure); } ``` ## Mapping Lifecycle Understanding how and when Elasticsearch field mappings are applied is critical to avoiding silent query failures. The behavior differs significantly by index type, and `DailyIndex`/`MonthlyIndex` require special attention. ### How Mappings Are Applied by Index Type | Index type | `ConfigureIndexesAsync` behavior | First write (without explicit configure) | How to apply a new field mapping to existing data | |---|---|---|---| | `Index` | Creates index if missing; calls PUT Mapping on existing index | `EnsureIndexAsync` triggers create-or-update (one-time, flag-guarded) | Automatic — `ConfigureIndexesAsync` or first write applies it | | `VersionedIndex` | Same as `Index`, targets the concrete versioned index (e.g., `employees-v2`) | Same one-time `EnsureIndexAsync` path | Automatic — same as `Index` | | `DailyIndex` | **No-op** — `ConfigureAsync` does nothing. Existing partitions are never updated. | Creates a new dated partition (with full mapping) only if one doesn't exist for that date | **Manual** — you must apply the mapping to existing partitions yourself (see below) | | `MonthlyIndex` | Same as `DailyIndex` | Same as `DailyIndex` | Same as `DailyIndex` | ::: warning DailyIndex and MonthlyIndex do not update existing partitions `DailyIndex.ConfigureAsync()` is intentionally a no-op. Neither `ConfigureIndexesAsync` nor the lazy `EnsureIndexAsync` path will ever call PUT Mapping on an already-created daily or monthly partition. Only **new** partitions created after you add the field mapping will have it. ::: ### What Happens Without Calling `ConfigureIndexesAsync` You are not required to call `ConfigureIndexesAsync` explicitly. Repository **write** operations (`AddAsync`, `SaveAsync`, `PatchAsync`, `RemoveAsync`, `PatchAllAsync`, `BatchProcessAsync`) call `EnsureIndexAsync` internally before mutating data. However, **read** operations (`FindAsync`, `GetByIdAsync`, `CountAsync`) do **not** call `EnsureIndexAsync`. If you query before any write has occurred, the index may not exist yet. ```mermaid flowchart TD subgraph entryPoints [Entry Points] ConfigureIndexesAsync["ConfigureIndexesAsync()"] FirstWrite["First repository write"] end ConfigureIndexesAsync --> PerIndex["For each index: ConfigureAsync()"] FirstWrite --> EnsureIndex["EnsureIndexAsync(target)"] PerIndex --> IndexT{"Index type?"} EnsureIndex --> IndexT2{"Index type?"} IndexT -->|"Index / VersionedIndex"| UpdateOrCreate["Create index if missing,\nor PUT Mapping if exists"] IndexT -->|"DailyIndex / MonthlyIndex"| NoOp["No-op (does nothing)"] IndexT2 -->|"Index / VersionedIndex"| OnceGuard["One-time: ConfigureAsync()\n(flag-guarded, includes PUT Mapping)"] IndexT2 -->|"DailyIndex / MonthlyIndex"| EnsureDate["EnsureDateIndexAsync:\nCreate partition if missing\n(full mapping on creation)"] ``` **For `Index` / `VersionedIndex`**: The first write auto-configures the index (create or update settings + mappings). It is safe to skip `ConfigureIndexesAsync` in development — the first mutation handles it. In production, calling `ConfigureIndexesAsync` on startup is still recommended to surface mapping errors early. **For `DailyIndex` / `MonthlyIndex`**: The first write to a new date creates that partition with the full current mapping. Writes to dates whose partitions already exist do nothing to the mapping. If you add a new field and only write to existing dates, the mapping is never applied anywhere. ### Updating Existing Daily/Monthly Partitions When you add a new field to `ConfigureIndexMapping` on a `DailyIndex` or `MonthlyIndex`, you have several options for existing partitions: | Strategy | Cost | When to use | |----------|------|-------------| | **Roll forward** (do nothing to old partitions) | Zero cost; new partitions pick up the mapping on creation | Feature can wait until enough data has naturally accumulated (e.g., after 7/30/90 days of retention). Best for non-critical analytics fields or gradual rollouts. | | **PutMapping + update-by-query on all partitions** | High I/O cost proportional to total data volume; re-indexes every document in every partition | Need the field searchable across all historical data immediately. Can saturate cluster I/O for hours. | | **Targeted backfill** (PutMapping + update-by-query on recent partitions only) | Moderate cost; only touches last N days/months | Need the field on recent data but older data will age out via retention anyway. | | **Bump version** (full reindex to new partitions) | Roughly same I/O cost as update-by-query but also doubles disk temporarily | Need a type change on an existing field, or you want a clean slate. | ::: tip Plan ahead to avoid backfill costs Add field mappings to `ConfigureIndexMapping` **early** — even before you write data to them. There is no cost to mapping a field you don't populate yet. This ensures all future partitions are ready when you start writing the field. ::: #### Practical Recommendations 1. **Roll forward by default.** For most analytics and reporting fields, add the mapping and wait. Once `MaxIndexAge` worth of partitions have been created with the new mapping, all queryable data will have it. 2. **Gate features on data availability.** If a UI feature depends on a new field, gate it on "created after deploy date" or gracefully handle missing data in older results. 3. **Factor retention into the decision.** If `MaxIndexAge` is 30 days and you can wait 30 days, you get full coverage for free without any backfill. 4. **Update-by-query is rarely worth it at scale.** For a `DailyIndex` with 90 days retention and millions of documents per day, an update-by-query touches the same total volume as a version bump reindex. The only advantage is no temporary disk doubling — but you still pay the full I/O cost. If you're paying that cost, consider whether a version bump gives you a cleaner outcome. 5. **Targeted backfill as a middle ground.** Apply PutMapping + update-by-query to only the last N days rather than full history. Example: ```bash # Apply mapping to all existing daily partitions PUT /logs-v1-*/_mapping { "properties": { "newField": { "type": "keyword" } } } # Re-index _source into the inverted index (no script needed) POST /logs-v1-2025.05.*/_update_by_query?conflicts=proceed ``` ### Mapping Resolver Cache (Query-Time Mapping Awareness) The repository framework does **not** cache the PUT Mapping request/response (that's purely server-side). However, the **query parser** uses an `ElasticMappingResolver` that caches field-to-type resolution for building queries, sorting, and aggregations. This resolver combines two sources: 1. **Code mapping** — derived from your `ConfigureIndexMapping` method at startup (immutable for the process lifetime) 2. **Server mapping** — fetched from the Elasticsearch GET Mapping API, cached in memory and **automatically refreshed at most once per minute** #### What this means after a manual PUT Mapping If you manually apply a mapping change (e.g., `PUT /index/_mapping` via the Elasticsearch API or a script), the `ElasticMappingResolver` will automatically pick it up within ~60 seconds on the next field resolution. You typically do not need to do anything in application code. If you need immediate recognition (e.g., in tests or a migration script that queries the new field right after applying the mapping), call: ```csharp index.MappingResolver.RefreshMapping(); ``` This clears the cached server mapping and forces the next `GetMapping()` call to re-fetch from the cluster. #### Cache lifetime summary | Cache layer | Lifetime | How to invalidate | |---|---|---| | `ElasticMappingResolver` field cache | Auto-refreshes from server every ~60 seconds | `index.MappingResolver.RefreshMapping()` | | `_isEnsured` flag (`Index` / `VersionedIndex`) | Process lifetime (one-time flag) | Deleting the index resets it; otherwise persists until app restart | | `_ensuredDates` (`DailyIndex`) | Process lifetime per-date | Cleared on `DeleteAsync(name)` or `Dispose()`; otherwise persists until app restart | | `ConfigureIndexesAsync` cache marker | 5 minutes (distributed via `ICacheClient`) | Automatically expires; or call `ConfigureIndexesAsync(force: true)` | #### No cluster-side action needed Elasticsearch itself has no mapping cache you need to invalidate — once a PUT Mapping succeeds, the mapping is immediately active for new indexing and queries. The only caching is in-process within the .NET application: * **For queries**: The `ElasticMappingResolver` auto-refreshes. If you need it sooner, call `RefreshMapping()`. * **For writes**: The `_isEnsured` / `_ensuredDates` flags only control whether `ConfigureAsync` runs again. They don't prevent writes to the index — they just skip redundant index creation/mapping calls. Manual PUT Mapping changes are orthogonal to these flags. ### In-Place Analysis Updates (analyzers, tokenizers, filters) For `Index` and `VersionedIndex`, adding new analysis components (analyzers, tokenizers, token filters, normalizers, char filters) to an existing index does **not** require a new index version. When `ConfigureIndexesAsync` re-runs against an existing index, the dynamic settings — including the `Analysis` block — are applied in place via a `PutSettings` call with `Reopen()`. The reopen briefly closes and reopens the index so the new components become active. This is unlike changing an existing **field mapping** type (which does require a new version on a `VersionedIndex`). Existing documents are not reindexed by an in-place analysis update, so a newly added analyzer only affects documents indexed (and queries run) after the upgrade. Before applying, the library diffs the desired analysis components against the live index and logs a `requires close/reopen` warning for each genuinely **new** component (see the table below). Components that already exist are not re-warned. ::: info Where Elasticsearch stores analysis settings: `Settings.Index.Analysis` vs root `Settings.Analysis` Elasticsearch exposes index analysis settings in two different shapes depending on direction: * **Reading** via the Get Settings API returns analysis nested under the `index` key — `Settings.Index.Analysis`. This is the canonical location for the **current** live state of an index. The root `Settings.Analysis` is **not** populated on reads. * **Writing** via a create/update request uses the root `Settings.Analysis` shape — the same shape your `ConfigureIndex(...).Analysis(...)` builder produces for the **desired** state. The in-place upgrade therefore compares the desired root `Settings.Analysis` (from `ConfigureIndex`) against the current `Settings.Index.Analysis` (from the Get Settings response). Reading the current set from the root `Settings.Analysis` would always return nothing, making every existing component look new and falsely warning on every upgrade. ::: ### Failure Log Messages When mapping or settings updates fail, the following log messages are emitted: | Level | Message | Meaning | |-------|---------|---------| | Error | `Error updating index ({name}) settings` | Index settings PUT failed | | Error | `Error updating index ({name}) mappings.` | PUT Mapping failed on `Index` | | Error | `Error updating index ({name}) mappings. Changing existing fields requires a new index version.` | PUT Mapping rejected on `VersionedIndex` — you tried to change an existing field's type | | Warning | `Adding new analyzer {AnalyzerKey} to existing index (requires close/reopen)` | New analyzer detected in settings; requires index close/reopen to take effect | | Warning | `Adding new tokenizer {TokenizerKey} to existing index (requires close/reopen)` | Same for tokenizers | | Warning | `Adding new token filter {TokenFilterKey} to existing index (requires close/reopen)` | Same for token filters | | Warning | `Adding new normalizer {NormalizerKey} to existing index (requires close/reopen)` | Same for normalizers | | Warning | `Adding new char filter {CharFilterKey} to existing index (requires close/reopen)` | Same for char filters | ::: info DailyIndex never emits mapping errors Since `DailyIndex.ConfigureAsync()` is a no-op, you will never see mapping error logs from the built-in configuration path for daily/monthly indexes. If a mapping is incompatible with an existing partition, you will only discover it when manually calling the PUT Mapping API. ::: ## Retention Policy for Time-Series Indexes ### Configuring Retention For `DailyIndex` and `MonthlyIndex`, configure retention with `MaxIndexAge`: ```csharp public sealed class LogEventIndex : DailyIndex { public LogEventIndex(IElasticConfiguration configuration) : base(configuration, "logs", version: 1) { // Keep indexes for 90 days MaxIndexAge = TimeSpan.FromDays(90); // Automatically delete expired indexes during maintenance DiscardExpiredIndexes = true; } } ``` ### How Retention Works 1. **Index Expiration**: Each index has an expiration date based on its date + `MaxIndexAge` 2. **Maintenance Job**: `MaintainIndexesAsync()` checks for expired indexes 3. **Automatic Deletion**: If `DiscardExpiredIndexes` is true, expired indexes are deleted ```csharp // Index: logs-v1-2024.01.15 // MaxIndexAge: 90 days // Expiration: 2024.01.15 + 90 days = 2024.04.15 // After 2024.04.15, this index is eligible for deletion ``` ### Running Maintenance Call `MaintainIndexesAsync()` regularly (e.g., via a scheduled job): ```csharp // In a background job public class IndexMaintenanceJob : IJob { private readonly MyElasticConfiguration _configuration; public async Task RunAsync(CancellationToken cancellationToken) { // This will: // 1. Update aliases for time-series indexes // 2. Delete expired indexes (if DiscardExpiredIndexes = true) await _configuration.MaintainIndexesAsync(); } } ``` Or use the built-in `MaintainIndexesJob`: ```csharp services.AddJob(o => o.ApplyDefaults()); ``` ### Preventing Writes to Expired Indexes The library prevents writing to indexes that have exceeded `MaxIndexAge`: ```csharp // If MaxIndexAge is 90 days and you try to write a document // with a date older than 90 days, an ArgumentException is thrown var oldDocument = new LogEvent { CreatedUtc = DateTime.UtcNow.AddDays(-100) // Older than MaxIndexAge }; // This will throw: "Index max age exceeded" await repository.AddAsync(oldDocument); ``` ### Time-Based Aliases Create aliases that automatically include only recent indexes: ```csharp public LogEventIndex(IElasticConfiguration configuration) : base(configuration, "logs", version: 1) { MaxIndexAge = TimeSpan.FromDays(90); DiscardExpiredIndexes = true; // Create aliases for recent data windows AddAlias("logs-last-7-days", TimeSpan.FromDays(7)); AddAlias("logs-last-30-days", TimeSpan.FromDays(30)); AddAlias("logs-last-90-days", TimeSpan.FromDays(90)); } ``` These aliases are automatically updated during maintenance: * `logs-last-7-days` only includes indexes from the last 7 days * Older indexes are removed from the alias but not deleted (until they exceed `MaxIndexAge`) ### Monthly Index Retention For `MonthlyIndex`, retention works the same way but with monthly granularity: ```csharp public sealed class AuditLogIndex : MonthlyIndex { public AuditLogIndex(IElasticConfiguration configuration) : base(configuration, "audit", version: 1) { // Keep audit logs for 1 year MaxIndexAge = TimeSpan.FromDays(365); DiscardExpiredIndexes = true; } } // Index naming: audit-v1-2024.01, audit-v1-2024.02, etc. // Expiration: End of month + 365 days ``` ### Retention Best Practices 1. **Set appropriate retention**: Balance storage costs with data retention requirements 2. **Run maintenance regularly**: Schedule `MaintainIndexesAsync()` daily or more frequently 3. **Monitor disk usage**: Track index sizes and adjust retention as needed 4. **Use aliases for queries**: Query against aliases like `logs-last-30-days` for better performance 5. **Consider compliance**: Ensure retention meets regulatory requirements ```csharp // Example: Different retention for different data types public class LogsIndex : DailyIndex { public LogsIndex(IElasticConfiguration config) : base(config, "logs") { MaxIndexAge = TimeSpan.FromDays(30); // Short retention for logs } } public class AuditIndex : MonthlyIndex { public AuditIndex(IElasticConfiguration config) : base(config, "audit") { MaxIndexAge = TimeSpan.FromDays(365 * 7); // 7 years for compliance } } ``` ## Index Operations ### Configure Indexes Create indexes and update mappings: ```csharp await configuration.ConfigureIndexesAsync(); ``` Options: * Creates indexes that don't exist * Updates mappings for existing indexes (if compatible) * Creates aliases * With the default `beginReindexingOutdated: true`, **enqueues** (does not run) a reindex work item for each outdated index — see [What actually triggers a reindex](#what-actually-triggers-a-reindex) for why you usually want `configuration.ReindexAsync()` instead #### Concurrency Protection When multiple distributed processes (pods, workers, migration runners) call `ConfigureIndexesAsync` on startup, a distributed lock and cache marker prevent redundant Elasticsearch admin API calls: 1. **Cache check**: If a configuration marker exists in the distributed cache, the call returns immediately with zero Elasticsearch calls and zero lock overhead. 2. **Distributed lock**: A distributed lock serializes concurrent callers so only one process runs the full configure pass at a time. 3. **Double-check**: After acquiring the lock, the cache is checked again in case another process finished while waiting. 4. **Configure**: The full configure and maintain pass runs on all indexes in parallel. 5. **Set cache marker**: A 5-minute TTL marker is set in the distributed cache so subsequent callers skip. The cache marker key includes a stable hash of all index names and versions, so deploying a new configuration (adding indexes, changing versions) automatically bypasses stale markers from a previous configuration. Old markers expire naturally after 5 minutes. The marker is explicitly cleared by `DeleteIndexesAsync` and `ReindexAsync` so the next configure call re-validates after any structural change. `MaintainIndexesAsync` does not clear the marker because it does not change index structure (names or versions). ```csharp // First call configures and sets the marker await configuration.ConfigureIndexesAsync(); // Subsequent calls within 5 minutes skip (fast path) await configuration.ConfigureIndexesAsync(); // Passing explicit indexes bypasses the lock and cache marker await configuration.ConfigureIndexesAsync([myIndex]); ``` ### Maintain Indexes Run maintenance tasks: ```csharp await configuration.MaintainIndexesAsync(); ``` Tasks: * Update aliases for time-series indexes * Delete expired indexes * Ensure index consistency ### Delete Indexes ```csharp // Delete all indexes await configuration.DeleteIndexesAsync(); // Delete specific index await index.DeleteAsync(); ``` ### Reindex ```csharp // Reindex all indexes await configuration.ReindexAsync(); // Reindex with progress callback await configuration.ReindexAsync(async (progress, message) => { Console.WriteLine($"{progress}%: {message}"); }); // Reindex specific index await index.ReindexAsync(); ``` ## Index Properties ### IIndex Interface ```csharp public interface IIndex : IDisposable { string Name { get; } bool HasMultipleIndexes { get; } IElasticQueryBuilder QueryBuilder { get; } ElasticMappingResolver MappingResolver { get; } ElasticQueryParser QueryParser { get; } IElasticConfiguration Configuration { get; } Task ConfigureAsync(); Task EnsureIndexAsync(object target); Task MaintainAsync(bool includeOptionalTasks = true); Task DeleteAsync(); Task ReindexAsync(Func progressCallbackAsync = null); string CreateDocumentId(object document); string[] GetIndexesByQuery(IRepositoryQuery query); string GetIndex(object target); } ``` ### Index Properties ```csharp public class Index { public string Name { get; } public bool HasMultipleIndexes { get; } public int BulkBatchSize { get; set; } = 1000; public int? ReindexBatchSize { get; set; } public float? ReindexRequestsPerSecond { get; set; } // Query field restrictions public ISet AllowedQueryFields { get; } public ISet AllowedAggregationFields { get; } public ISet AllowedSortFields { get; } } ``` ### VersionedIndex Properties ```csharp public class VersionedIndex { public int Version { get; } public string VersionedName { get; } // e.g., "employees-v2" public bool DiscardIndexesOnReindex { get; set; } } ``` ### DailyIndex Properties ```csharp public class DailyIndex { public TimeSpan? MaxIndexAge { get; set; } public bool DiscardExpiredIndexes { get; set; } } ``` ## Best Practices ### 1. Use Versioned Indexes for Evolving Schemas ```csharp // Start with version 1 public EmployeeIndex(...) : base(configuration, "employees", version: 1) { } // Increment when schema changes public EmployeeIndex(...) : base(configuration, "employees", version: 2) { } ``` ### 2. Use Time-Series Indexes for Log Data ```csharp // Daily for high-volume, short retention public class LogIndex : DailyIndex { } // Monthly for lower-volume, longer retention public class AuditIndex : MonthlyIndex { } ``` ### 3. Configure Appropriate Retention ```csharp MaxIndexAge = TimeSpan.FromDays(90); DiscardExpiredIndexes = true; ``` ### 4. Use Aliases for Zero-Downtime Migrations ```csharp // Alias always points to current version // Applications use alias, not versioned index name ``` ### 5. Test Reindex Scripts ```csharp // Test scripts in development before production AddReindexScript(2, @" // Validate script works correctly ctx._source.newField = ctx._source.oldField; "); ``` ## Next Steps * [Migrations](/guide/migrations) - Document migrations * [Jobs](/guide/jobs) - Index maintenance jobs * [Elasticsearch Setup](/guide/elasticsearch-setup) - Connection configuration ## Concurrency Safety Reindexing is protected by a distributed lock keyed on the index alias to prevent concurrent reindex operations from corrupting data. ### Lock Strategy * **Lock key**: `reindex:{alias}` (e.g., `reindex:employees`) * **Lock TTL**: 20 minutes, auto-renewed during long-running operations * Both direct (`VersionedIndex.ReindexAsync`) and work-item (`ReindexWorkItemHandler`) paths use the same lock * Only one reindex per logical index can run at a time — subsequent version transitions wait for the current one to complete ### Why Alias-Only Keys Using the alias as the lock key ensures that sequential version transitions (v1→v2, then v2→v3) cannot overlap. If v2→v3 started before v1→v2 completed, v3 would contain incomplete data from v2. ### Lock Renewal for Long-Running Reindexes For indexes with millions of documents that take hours to reindex, the lock is automatically renewed on every progress callback (every 1-10 seconds during the polling loop). This prevents lock expiration during legitimate long-running operations. ### Crash Recovery If an instance crashes mid-reindex, the lock expires after 20 minutes. Another instance can then retry the reindex. `VersionedIndex.ReindexAsync()` is resume-safe — it picks up from the last document using timestamp-based or ID-based range queries. ### Second-Pass Catch-Up Strategy Reindexing performs a second pass after the first completes to catch documents written during the first pass. The strategy depends on the index configuration: 1. **TimestampField available** (e.g., `IHaveDates` models): Uses a timestamp-based range query starting from the reindex start time. This is the preferred approach. 2. **No TimestampField, ObjectId-format IDs**: Falls back to ObjectId-based range queries on the document `id` field (ObjectIds encode a timestamp). Logged at Information level. 3. **No TimestampField, non-ObjectId IDs**: Cannot perform a second pass. Logs a Warning — documents written during reindex may be lost. Consider adding `IHaveDates` to your model or using ObjectId-format IDs. 4. **Empty source index**: Skips the second pass entirely (nothing to catch up). ### Unique Index Names `ElasticConfiguration.AddIndex()` enforces unique index names (case-insensitive). Registering two indexes with the same alias throws an `ArgumentException` at startup, preventing conflicts before they can cause data corruption. --- --- url: /guide/migrations.md --- # Migrations Foundatio.Repositories provides a migration system for evolving your data schema over time. This guide covers creating and running migrations. ## Overview Migrations allow you to: * Transform existing documents when schema changes * Run one-time data fixes * Perform repeatable maintenance tasks * Track migration history ## Migration Types ### MigrationType Enum ```csharp public enum MigrationType { Versioned, // Run once, sequential, fails permanently on error VersionedAndResumable, // Run once, sequential, auto-retry on failure Repeatable // Run after versioned migrations, can run multiple times } ``` | Type | Runs | On Failure | Use Case | |------|------|------------|----------| | `Versioned` | Once | Fails permanently | Schema changes | | `VersionedAndResumable` | Once | Auto-retry | Long-running migrations | | `Repeatable` | Multiple times | Retry | Maintenance tasks | ## Creating Migrations ### IMigration Interface ```csharp public interface IMigration { MigrationType MigrationType { get; } int? Version { get; } bool RequiresOffline { get; } Task RunAsync(MigrationContext context); } ``` ### Basic Migration ```csharp public class AddDepartmentFieldMigration : IMigration { private readonly IEmployeeRepository _repository; public AddDepartmentFieldMigration(IEmployeeRepository repository) { _repository = repository; } public MigrationType MigrationType => MigrationType.Versioned; public int? Version => 1; public bool RequiresOffline => false; public async Task RunAsync(MigrationContext context) { await _repository.PatchAllAsync( q => q.FieldEmpty(e => e.Department), new PartialPatch(new { Department = "General" })); } } ``` ### Using MigrationBase ```csharp public class AddDepartmentFieldMigration : MigrationBase { private readonly IEmployeeRepository _repository; public AddDepartmentFieldMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.Versioned; } public override int? Version => 1; public override async Task RunAsync(MigrationContext context) { await _repository.PatchAllAsync( q => q.FieldEmpty(e => e.Department), new PartialPatch(new { Department = "General" })); _logger.LogInformation("Added default department to employees"); } } ``` ### Resumable Migration For long-running migrations that should resume on failure: ```csharp public class BackfillDataMigration : MigrationBase { private readonly IEmployeeRepository _repository; public BackfillDataMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.VersionedAndResumable; } public override int? Version => 2; public override async Task RunAsync(MigrationContext context) { long processed = 0; await _repository.BatchProcessAsync( q => q.FieldEmpty(e => e.CalculatedField), async batch => { foreach (var employee in batch.Documents) { employee.CalculatedField = CalculateValue(employee); } await _repository.SaveAsync(batch.Documents); processed += batch.Documents.Count; _logger.LogInformation("Processed {Count} employees", processed); return true; // Continue processing }, o => o.PageLimit(100)); } } ``` ### Repeatable Migration For tasks that should run periodically: ```csharp public class CleanupExpiredDataMigration : MigrationBase { private readonly IEmployeeRepository _repository; public CleanupExpiredDataMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.Repeatable; } public override int? Version => null; // No version for repeatable public override async Task RunAsync(MigrationContext context) { var cutoffDate = DateTime.UtcNow.AddYears(-7); var deleted = await _repository.RemoveAllAsync( q => q.DateRange(null, cutoffDate, e => e.TerminationDate), o => o.IncludeSoftDeletes()); _logger.LogInformation("Deleted {Count} expired employee records", deleted); } } ``` ## Running Migrations ### MigrationManager ```csharp var manager = new MigrationManager( serviceProvider, migrationStateRepository, lockProvider, loggerFactory); // Register migrations from assembly manager.AddMigrationsFromAssembly(); // Run all pending migrations var result = await manager.RunMigrationsAsync(); Console.WriteLine($"Success: {result == MigrationResult.Success}"); ``` ### Migration Result ```csharp public enum MigrationResult { Success, Failed, UnableToAcquireLock, Cancelled } ``` ### Dependency Injection Setup ```csharp // Register migration services services.AddSingleton(); services.AddSingleton(new InMemoryLockProvider()); // Register migrations services.AddTransient(); services.AddTransient(); // Register migration manager services.AddSingleton(sp => { var manager = new MigrationManager( sp, sp.GetRequiredService(), sp.GetRequiredService(), sp.GetRequiredService()); manager.AddMigrationsFromAssembly(); return manager; }); ``` ### Running on Startup ```csharp public class MigrationStartupAction : IStartupAction { private readonly MigrationManager _migrationManager; private readonly ILogger _logger; public MigrationStartupAction(MigrationManager migrationManager, ILogger logger) { _migrationManager = migrationManager; _logger = logger; } public async Task RunAsync(CancellationToken cancellationToken = default) { _logger.LogInformation("Running migrations..."); var result = await _migrationManager.RunMigrationsAsync(); if (result != MigrationResult.Success) { _logger.LogError("Migration failed: {Result}", result); throw new InvalidOperationException($"Migration failed: {result}"); } _logger.LogInformation("Migrations completed successfully"); } } // Register startup action services.AddStartupAction(); ``` ## Migration Context The `MigrationContext` provides access to the migration lock, logger, and cancellation: ```csharp public class MigrationContext { public ILock Lock { get; } public ILogger Logger { get; } public CancellationToken CancellationToken { get; } } ``` ### Renewing Locks for Long-Running Migrations For long-running migrations, you should periodically renew the lock to prevent it from expiring: ```csharp public class LongRunningMigration : MigrationBase { private readonly IEmployeeRepository _repository; public LongRunningMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.VersionedAndResumable; } public override int? Version => 10; public override async Task RunAsync(MigrationContext context) { long processed = 0; await _repository.BatchProcessAsync( q => q.All(), async batch => { // Process batch foreach (var employee in batch.Documents) { employee.CalculatedField = CalculateValue(employee); } await _repository.SaveAsync(batch.Documents, o => o.Notifications(false)); processed += batch.Documents.Count; // Renew lock every batch to prevent expiration await context.Lock.RenewAsync(TimeSpan.FromMinutes(30)); context.Logger.LogInformation("Processed {Count} employees, lock renewed", processed); // Check for cancellation return !context.CancellationToken.IsCancellationRequested; }, o => o.PageLimit(500)); } } ``` ::: tip Lock Renewal Best Practices * Renew the lock at regular intervals (e.g., every batch or every few minutes) * The `MigrationManager` automatically renews locks for `VersionedAndResumable` migrations before each retry * For `Versioned` migrations, you must manually renew if the migration takes longer than the lock timeout (30 minutes by default) ::: ### Using the Logger The context provides a logger scoped to your migration class: ```csharp public override async Task RunAsync(MigrationContext context) { context.Logger.LogInformation("Starting migration..."); try { await DoMigrationWorkAsync(); context.Logger.LogInformation("Migration completed successfully"); } catch (Exception ex) { context.Logger.LogError(ex, "Migration failed"); throw; } } ``` ### Handling Cancellation ```csharp public override async Task RunAsync(MigrationContext context) { await _repository.BatchProcessAsync( query, async batch => { // Check for cancellation context.CancellationToken.ThrowIfCancellationRequested(); // Process batch... return true; }); } ``` ## Migration Patterns ### Schema Migration ```csharp public class RenameFieldMigration : MigrationBase { private readonly IEmployeeRepository _repository; public RenameFieldMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.Versioned; } public override int? Version => 3; public override async Task RunAsync(MigrationContext context) { // Copy old field to new field await _repository.PatchAllAsync( q => q.FieldHasValue(e => e.OldFieldName), new ScriptPatch(@" ctx._source.newFieldName = ctx._source.oldFieldName; ctx._source.remove('oldFieldName'); ")); } } ``` ### Data Transformation ```csharp public class NormalizeEmailMigration : MigrationBase { private readonly IEmployeeRepository _repository; public NormalizeEmailMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.Versioned; } public override int? Version => 4; public override async Task RunAsync(MigrationContext context) { await _repository.PatchAllAsync( q => q.FieldHasValue(e => e.Email), new ScriptPatch("ctx._source.email = ctx._source.email.toLowerCase()")); } } ``` ### Backfill Calculated Fields ```csharp public class BackfillFullNameMigration : MigrationBase { private readonly IEmployeeRepository _repository; public BackfillFullNameMigration(IEmployeeRepository repository) { _repository = repository; MigrationType = MigrationType.VersionedAndResumable; } public override int? Version => 5; public override async Task RunAsync(MigrationContext context) { await _repository.BatchProcessAsync( q => q.FieldEmpty(e => e.FullName), async batch => { foreach (var emp in batch.Documents) { emp.FullName = $"{emp.FirstName} {emp.LastName}"; } await _repository.SaveAsync(batch.Documents, o => o.Notifications(false)); return true; }, o => o.PageLimit(500)); } } ``` ### Data Cleanup ```csharp public class RemoveOrphanedRecordsMigration : MigrationBase { private readonly IEmployeeRepository _employeeRepo; private readonly ICompanyRepository _companyRepo; public RemoveOrphanedRecordsMigration( IEmployeeRepository employeeRepo, ICompanyRepository companyRepo) { _employeeRepo = employeeRepo; _companyRepo = companyRepo; MigrationType = MigrationType.Repeatable; } public override async Task RunAsync(MigrationContext context) { // Find all company IDs var companies = await _companyRepo.GetAllAsync(); var validCompanyIds = companies.Documents.Select(c => c.Id).ToHashSet(); // Remove employees with invalid company IDs await _employeeRepo.BatchProcessAsync( q => q.All(), async batch => { var orphaned = batch.Documents .Where(e => !validCompanyIds.Contains(e.CompanyId)) .ToList(); if (orphaned.Any()) { await _employeeRepo.RemoveAsync(orphaned); _logger.LogInformation("Removed {Count} orphaned employees", orphaned.Count); } return true; }); } } ``` ## Best Practices ### 1. Make Migrations Idempotent ```csharp // Good: Check before modifying await repository.PatchAllAsync( q => q.FieldEmpty(e => e.NewField), // Only update if not set new PartialPatch(new { NewField = "default" })); // Bad: Always update await repository.PatchAllAsync( q => q.All(), new PartialPatch(new { NewField = "default" })); ``` ### 2. Use Batch Processing for Large Datasets ```csharp await repository.BatchProcessAsync( query, async batch => { // Process in batches to avoid memory issues return true; }, o => o.PageLimit(500)); ``` ### 3. Disable Notifications for Bulk Updates ```csharp await repository.SaveAsync(documents, o => o.Notifications(false)); ``` ### 4. Log Progress ```csharp long processed = 0; await repository.BatchProcessAsync(query, async batch => { processed += batch.Documents.Count; _logger.LogInformation("Processed {Count} of {Total}", processed, batch.Total); return true; }); ``` ### 5. Test Migrations ```csharp [Fact] public async Task Migration_Should_Update_Department() { // Arrange var employee = await _repository.AddAsync(new Employee { Name = "Test" }); var lock_ = await _lockProvider.AcquireAsync("test"); // Act var migration = new AddDepartmentFieldMigration(_repository); await migration.RunAsync(new MigrationContext(lock_, _logger, CancellationToken.None)); // Assert var updated = await _repository.GetByIdAsync(employee.Id); Assert.Equal("General", updated.Department); } ``` ## Next Steps * [Index Management](/guide/index-management) - Schema versioning * [Jobs](/guide/jobs) - Scheduled maintenance * [Configuration](/guide/configuration) - Migration configuration --- --- url: /guide/jobs.md --- # Jobs Foundatio.Repositories provides built-in jobs for index maintenance, snapshots, and cleanup. This guide covers the available jobs and how to use them. ## Overview Jobs are background tasks that perform maintenance operations on your Elasticsearch indexes. They're built on Foundatio's job infrastructure. ## Available Jobs ### MaintainIndexesJob Runs maintenance tasks on all configured indexes: ```csharp public class MaintainIndexesJob : IJob { private readonly IElasticConfiguration _configuration; public MaintainIndexesJob(IElasticConfiguration configuration) { _configuration = configuration; } public async Task RunAsync(CancellationToken cancellationToken = default) { await _configuration.MaintainIndexesAsync(); return JobResult.Success; } } ``` **Tasks performed:** * Update aliases for time-series indexes * Delete expired indexes (if `DiscardExpiredIndexes` is true) * Ensure index consistency **Usage:** ```csharp // Register the job services.AddJob(); // Run manually var job = new MaintainIndexesJob(configuration); await job.RunAsync(); ``` ### SnapshotJob Creates Elasticsearch snapshots for backup: ```csharp public class SnapshotJob : IJob { private readonly ElasticsearchClient _client; private readonly string _repositoryName; public SnapshotJob(ElasticsearchClient client, string repositoryName = "backups") { _client = client; _repositoryName = repositoryName; } public async Task RunAsync(CancellationToken cancellationToken = default) { var snapshotName = $"snapshot-{DateTime.UtcNow:yyyy-MM-dd-HH-mm-ss}"; var response = await _client.Snapshot.CreateAsync( _repositoryName, snapshotName, s => s.WaitForCompletion(false)); if (!response.IsValidResponse) return JobResult.FromException(response.OriginalException(), response.GetErrorMessage()); return JobResult.Success; } } ``` **Configuration:** ```csharp // Register snapshot repository in Elasticsearch first await client.Snapshot.CreateRepositoryAsync("backups", r => r .FileSystem(fs => fs .Location("/mnt/backups") .Compress(true))); // Then use the job var job = new SnapshotJob(client, "backups"); await job.RunAsync(); ``` ### CleanupSnapshotJob Cleans up old snapshots: ```csharp public class CleanupSnapshotJob : IJob { private readonly ElasticsearchClient _client; private readonly string _repositoryName; private readonly TimeSpan _maxAge; public CleanupSnapshotJob(ElasticsearchClient client, string repositoryName, TimeSpan maxAge) { _client = client; _repositoryName = repositoryName; _maxAge = maxAge; } public async Task RunAsync(CancellationToken cancellationToken = default) { var cutoffDate = DateTime.UtcNow - _maxAge; var snapshots = await _client.Snapshot.GetAsync(_repositoryName, "_all"); foreach (var snapshot in snapshots.Snapshots) { if (snapshot.StartTime < cutoffDate) { await _client.Snapshot.DeleteAsync(_repositoryName, snapshot.Name); } } return JobResult.Success; } } ``` ### CleanupIndexesJob Deletes old indexes based on patterns and age: ```csharp public class CleanupIndexesJob : IJob { private readonly ElasticsearchClient _client; private readonly string _indexPattern; private readonly TimeSpan _maxAge; public CleanupIndexesJob(ElasticsearchClient client, string indexPattern, TimeSpan maxAge) { _client = client; _indexPattern = indexPattern; _maxAge = maxAge; } public async Task RunAsync(CancellationToken cancellationToken = default) { var cutoffDate = DateTime.UtcNow - _maxAge; var indices = await _client.Cat.IndicesAsync(i => i.Index(_indexPattern)); foreach (var index in indices.Records) { // Parse date from index name (e.g., logs-2024.01.15) if (TryParseIndexDate(index.Index, out var indexDate) && indexDate < cutoffDate) { await _client.Indices.DeleteAsync(index.Index); } } return JobResult.Success; } } ``` ### ElasticMigrationJob `ElasticMigrationJobBase` is an **abstract, opt-in** base class for a job that runs pending data migrations and then reindexes every outdated versioned index: ```csharp public abstract class ElasticMigrationJobBase : JobBase { protected override async Task RunInternalAsync(JobContext context) { // Create/update indexes and mappings without enqueuing the (no-op for // time-series) queue-based reindex path: await _configuration.ConfigureIndexesAsync(null, false); await _migrationManager.Value.RunMigrationsAsync(); // Reindexes every outdated IVersionedIndex directly (index.ReindexAsync()), // one index at a time internally, but all outdated indexes IN PARALLEL // (Task.WhenAll) relative to each other. var tasks = _configuration.Indexes.OfType().Select(ReindexIfNecessary); await Task.WhenAll(tasks); return JobResult.Success; } } ``` ::: warning This is scaffolding you opt into — nothing runs it for you `ElasticMigrationJobBase` is **not** auto-registered, auto-discovered, or scheduled by the library, and nothing in this repo (including its own [sample app](https://github.com/FoundatioFx/Foundatio.Repositories/tree/main/samples/Foundatio.SampleApp)) derives from it. If you want a repeatable "run migrations + reindex everything outdated" job, derive from it yourself and register/schedule it in your own app (`services.AddJob()`). For a one-time version upgrade, it's simpler — and more predictable — to call `configuration.ReindexAsync()` / `index.ReindexAsync()` directly; see [What actually triggers a reindex](/guide/index-management#what-actually-triggers-a-reindex). ::: **Usage (if you choose to derive from it):** ```csharp public class MyMigrationJob : ElasticMigrationJobBase { public MyMigrationJob(MigrationManager migrationManager, IElasticConfiguration configuration, ILoggerFactory loggerFactory) : base(migrationManager, configuration, loggerFactory) { } protected override void Configure(MigrationManager manager) { // Register your MigrationBase implementations here } } // Register and run it like any other job services.AddJob(); ``` ### ReindexWorkItemHandler Handles reindexing operations as background work items with automatic lock renewal and progress reporting: ```csharp public class ReindexWorkItem { public string OldIndex { get; set; } public string NewIndex { get; set; } public string Alias { get; set; } public string Script { get; set; } // Painless script for data transformation public bool DeleteOld { get; set; } // Delete old index after successful reindex public string TimestampField { get; set; } // Field for incremental reindex public DateTime? StartUtc { get; set; } // Start time for incremental reindex public int? ReindexBatchSize { get; set; } // Documents per internal bulk batch (default: 1000) public float? ReindexRequestsPerSecond { get; set; } // Throttle in docs/sec (default: unlimited) } ``` **Features:** * **Automatic Lock Renewal**: The handler sets `AutoRenewLockOnProgress = true`, which automatically renews the distributed lock whenever progress is reported * **Progress Reporting**: Reports progress percentage and status messages during reindex * **Two-Pass Reindex**: Performs a second pass to catch documents modified during the first pass. Uses `TimestampField` if available; falls back to ObjectId-based range queries if document IDs are ObjectId-format; logs a warning if neither strategy is available * **Error Handling**: Failed documents are stored in an error index (`{newIndex}-error`) * **Resilient Status Polling**: Progress is tracked by repeatedly polling the Elasticsearch task status API; failures back off exponentially with jitter (1 second, doubling up to a 30 second cap, +/-25% jitter) instead of retrying immediately, so a struggling cluster isn't hammered with repeated requests, and multiple work items failing at once don't retry in lockstep * **Stall Detection Scales With Throttle**: A reindex making no progress for too long is treated as stalled and abandoned. The threshold defaults to 10 minutes, but when `ReindexRequestsPerSecond` is set low enough that Elasticsearch's own inter-batch pause (`ReindexBatchSize` ÷ `ReindexRequestsPerSecond`, with a 3x safety margin) would exceed 10 minutes, the threshold extends to cover it - so a healthy, intentionally throttled reindex isn't mistaken for a stalled one * **Validated Throttle Settings**: `ReindexBatchSize`/`ReindexRequestsPerSecond` must be positive, finite numbers when set - `ReindexAsync` throws `ArgumentOutOfRangeException` immediately for a zero, negative, `NaN`, or infinite value ::: warning Enqueuing a work item is not enough on its own For this to run, something has to actually dequeue `ReindexWorkItem`s and dispatch them to `ReindexWorkItemHandler` — you need your own queue worker with the handler registered (e.g. via a `JobManager`/`WorkItemHandlers` setup). Simply calling `queue.EnqueueAsync(...)` below, with nothing processing the queue, leaves the work item sitting there indefinitely. Also, **this path does not work for `DailyIndex`/`MonthlyIndex`** — the old/new index names must be exact, existing index (or alias) names, so a dated partition's real name (`audit-v1-2024.01`) has to be used, not the unversioned base name. See [What actually triggers a reindex](/guide/index-management#what-actually-triggers-a-reindex) for the recommended direct alternative for time-series indexes. ::: **Usage:** ```csharp // Queue a reindex work item await queue.EnqueueAsync(new ReindexWorkItem { OldIndex = "employees-v1", NewIndex = "employees-v2", Alias = "employees", Script = "ctx._source.department = ctx._source.dept; ctx._source.remove('dept');", DeleteOld = true, TimestampField = "updatedUtc", // Enable two-pass reindex ReindexBatchSize = 200, // Lower this if large documents trip indexing pressure limits ReindexRequestsPerSecond = 500 // Optional: throttle to reduce load on a busy cluster }); ``` See [Throttling Reindex Load](/guide/index-management#throttling-reindex-load) and [Reindex Rejected Due to Indexing Pressure](/guide/troubleshooting#reindex-rejected-due-to-indexing-pressure) for why you'd set these. ## Reindex Progress Monitoring The `ElasticReindexer` provides detailed progress reporting during reindex operations: ### Progress Callback ```csharp await configuration.ReindexAsync(async (progress, message) => { // progress: 0-100 percentage // message: Status description _logger.LogInformation("Reindex {Progress}%: {Message}", progress, message); // Update metrics or UI await UpdateProgressMetricAsync(progress); }); ``` ### Progress Stages The reindex process reports progress through several stages: | Progress | Stage | |----------|-------| | 0% | Starting reindex | | 0-90% | First pass: copying documents | | 91% | First pass complete, updating aliases | | 92% | Aliases updated | | 92-96% | Second pass: catching modified documents | | 97% | Second pass complete | | 98% | Verifying document counts | | 99% | Deleting old index (if configured) | | 100% | Complete | ### Progress Messages Example progress messages during reindex: ``` 0%: Starting reindex... 45%: Total: 1,000,000 Completed: 450,000 VersionConflicts: 0 90%: Total: 1,000,000 Completed: 900,000 VersionConflicts: 12 91%: Total: 1,000,000 Completed: 1,000,000 92%: Updated aliases: employees Remove: employees-v1 Add: employees-v2 97%: Total: 150 Completed: 150 (second pass) 98%: Old Docs: 1,000,000 New Docs: 1,000,012 99%: Deleted index: employees-v1 100%: Complete ``` ### Monitoring Reindex in Migrations When using reindex within a migration, combine progress reporting with lock renewal: ```csharp public class ReindexMigration : MigrationBase { private readonly IElasticConfiguration _configuration; public ReindexMigration(IElasticConfiguration configuration) { _configuration = configuration; } public override MigrationType MigrationType => MigrationType.VersionedAndResumable; public override int? Version => 15; public override async Task RunAsync(MigrationContext context) { await _configuration.ReindexAsync(async (progress, message) => { context.Logger.LogInformation("Reindex {Progress}%: {Message}", progress, message); // Renew lock during long reindex operations await context.Lock.RenewAsync(TimeSpan.FromMinutes(30)); }); } } ``` ### Handling Reindex Failures Failed documents are stored in an error index for investigation: ```csharp // After reindex, check for failures var errorIndex = "employees-v2-error"; var existsResponse = await client.Indices.ExistsAsync(errorIndex); if (existsResponse.Exists) { var failures = await client.SearchAsync(s => s .Index(errorIndex) .Size(100)); foreach (var hit in failures.Hits) { _logger.LogError("Failed to reindex document: {Id}", hit.Id); // hit.Source contains: Index, Id, Version, Routing, Source, Cause, Status, Found } } ``` ## Running Jobs ### Manual Execution ```csharp var job = new MaintainIndexesJob(configuration); var result = await job.RunAsync(); if (result.IsSuccess) { Console.WriteLine("Maintenance completed"); } else { Console.WriteLine($"Maintenance failed: {result.Error}"); } ``` ### Scheduled Execution Using Foundatio's job runner: ```csharp // Run job on a schedule var runner = new JobRunner( new MaintainIndexesJob(configuration), loggerFactory, runContinuous: true, interval: TimeSpan.FromHours(1)); await runner.RunAsync(); ``` ### Hosted Service ```csharp public class MaintenanceHostedService : BackgroundService { private readonly IElasticConfiguration _configuration; private readonly ILogger _logger; public MaintenanceHostedService( IElasticConfiguration configuration, ILogger logger) { _configuration = configuration; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { while (!stoppingToken.IsCancellationRequested) { try { _logger.LogInformation("Running index maintenance..."); await _configuration.MaintainIndexesAsync(); _logger.LogInformation("Index maintenance completed"); } catch (Exception ex) { _logger.LogError(ex, "Index maintenance failed"); } await Task.Delay(TimeSpan.FromHours(1), stoppingToken); } } } // Register services.AddHostedService(); ``` ### Cron-Based Scheduling ```csharp public class ScheduledMaintenanceJob : IJob { private readonly IElasticConfiguration _configuration; public ScheduledMaintenanceJob(IElasticConfiguration configuration) { _configuration = configuration; } public async Task RunAsync(CancellationToken cancellationToken = default) { await _configuration.MaintainIndexesAsync(); return JobResult.Success; } } // Register with cron schedule (using Foundatio.Jobs.Hosting) services.AddCronJob("0 0 * * *"); // Daily at midnight ``` ## Custom Jobs ### Index Statistics Job ```csharp public class IndexStatisticsJob : IJob { private readonly ElasticsearchClient _client; private readonly ILogger _logger; public IndexStatisticsJob(ElasticsearchClient client, ILogger logger) { _client = client; _logger = logger; } public async Task RunAsync(CancellationToken cancellationToken = default) { var stats = await _client.Indices.StatsAsync("_all"); foreach (var index in stats.Indices) { _logger.LogInformation( "Index {Name}: {Docs} docs, {Size}", index.Key, index.Value.Primaries.Documents.Count, index.Value.Primaries.Store.Size); } return JobResult.Success; } } ``` ### Health Check Job ```csharp public class ElasticsearchHealthJob : IJob { private readonly ElasticsearchClient _client; private readonly ILogger _logger; public ElasticsearchHealthJob(ElasticsearchClient client, ILogger logger) { _client = client; _logger = logger; } public async Task RunAsync(CancellationToken cancellationToken = default) { var health = await _client.Cluster.HealthAsync(); _logger.LogInformation( "Cluster health: {Status}, Nodes: {Nodes}, Shards: {Shards}", health.Status, health.NumberOfNodes, health.ActiveShards); if (health.Status == HealthStatus.Red) { _logger.LogError("Cluster is in RED status!"); return JobResult.FromException(new Exception("Cluster health is RED")); } return JobResult.Success; } } ``` ### Data Archival Job ```csharp public class ArchiveOldDataJob : IJob { private readonly IEmployeeRepository _repository; private readonly IArchiveRepository _archiveRepository; private readonly ILogger _logger; public ArchiveOldDataJob( IEmployeeRepository repository, IArchiveRepository archiveRepository, ILogger logger) { _repository = repository; _archiveRepository = archiveRepository; _logger = logger; } public async Task RunAsync(CancellationToken cancellationToken = default) { var cutoffDate = DateTime.UtcNow.AddYears(-5); long archived = 0; await _repository.BatchProcessAsync( q => q.DateRange(null, cutoffDate, (Employee e) => e.TerminationDate), async batch => { // Archive to cold storage await _archiveRepository.AddAsync(batch.Documents); // Remove from hot storage await _repository.RemoveAsync(batch.Documents); archived += batch.Documents.Count; _logger.LogInformation("Archived {Count} records", archived); return !cancellationToken.IsCancellationRequested; }, o => o.IncludeSoftDeletes()); _logger.LogInformation("Archival completed: {Total} records archived", archived); return JobResult.Success; } } ``` ## Job Patterns ### Retry with Backoff ```csharp public async Task RunAsync(CancellationToken cancellationToken = default) { int retries = 3; TimeSpan delay = TimeSpan.FromSeconds(1); while (retries > 0) { try { await DoWorkAsync(cancellationToken); return JobResult.Success; } catch (Exception ex) when (retries > 1) { _logger.LogWarning(ex, "Job failed, retrying in {Delay}...", delay); await Task.Delay(delay, cancellationToken); delay *= 2; // Exponential backoff retries--; } } return JobResult.FromException(new Exception("Job failed after retries")); } ``` ### Distributed Locking ```csharp public async Task RunAsync(CancellationToken cancellationToken = default) { await using var lockHandle = await _lockProvider.AcquireAsync( "maintenance-job", TimeSpan.FromMinutes(30), cancellationToken); if (lockHandle == null) { _logger.LogInformation("Could not acquire lock, another instance is running"); return JobResult.Success; } await DoMaintenanceAsync(cancellationToken); return JobResult.Success; } ``` ### Progress Reporting ```csharp public async Task RunAsync(CancellationToken cancellationToken = default) { long total = await _repository.CountAsync(); long processed = 0; await _repository.BatchProcessAsync( q => q.All(), async batch => { await ProcessBatchAsync(batch); processed += batch.Documents.Count; var progress = (double)processed / total * 100; _logger.LogInformation("Progress: {Progress:F1}%", progress); return true; }); return JobResult.Success; } ``` ## Best Practices ### 1. Use Distributed Locks ```csharp await using var lockHandle = await _lockProvider.AcquireAsync("job-name"); if (lockHandle == null) return JobResult.Success; ``` ### 2. Handle Cancellation ```csharp while (!cancellationToken.IsCancellationRequested) { await ProcessNextBatchAsync(); } ``` ### 3. Log Progress ```csharp _logger.LogInformation("Processed {Count} of {Total}", processed, total); ``` ### 4. Use Appropriate Intervals ```csharp // Maintenance: hourly or daily // Snapshots: daily // Cleanup: weekly ``` ### 5. Monitor Job Health ```csharp if (!result.IsSuccess) { _metrics.IncrementCounter("job_failures", new { job = "maintenance" }); } ``` ## Next Steps * [Index Management](/guide/index-management) - Index configuration * [Migrations](/guide/migrations) - Data migrations * [Configuration](/guide/configuration) - Job configuration --- --- url: /guide/custom-fields.md --- # Custom Fields Foundatio.Repositories supports dynamic custom fields for tenant-specific or user-defined data. This guide covers setup, usage, lifecycle management, and best practices. ## Overview Custom fields allow you to: * Add tenant-specific fields without schema changes * Support user-defined attributes * Index dynamic data with proper field types * Query and aggregate on custom fields ## The Elasticsearch Field Limit Problem ### Why Custom Fields Matter Elasticsearch enforces a **default limit of 1,000 fields per index** through the `index.mapping.total_fields.limit` setting. This limit exists to prevent "mapping explosion" — uncontrolled field growth that causes: * **High memory pressure**: Each field mapping consumes JVM heap memory * **Slow cluster startup**: Large mappings take longer to load * **Performance degradation**: Query planning becomes slower with more fields * **Index corruption risk**: Extremely large mappings can cause stability issues When you exceed this limit, Elasticsearch rejects document ingestion with: ``` Limit of total fields [1000] has been exceeded while adding new fields ``` ### What Counts Toward the Limit The limit counts **all mappers**, not just leaf fields: | Type | Example | Count | |------|---------|-------| | Object mappers | `employee.address.city` | 3 (employee, address, city) | | Field mappers | `name`, `email`, `age` | 1 each | | Multi-fields | `name.keyword`, `name.raw` | 1 each | | Field aliases | Any alias | 1 each | ### The Multi-Tenant Challenge In multi-tenant applications, each tenant may want their own custom fields: ``` Tenant A: customField1, customField2, customField3 Tenant B: departmentCode, region, priority Tenant C: projectId, clientRef, billingCode ... ``` With 100 tenants each wanting 10 custom fields, you'd need 1,000 fields just for custom data — hitting the limit immediately. ### Naive Solutions (And Why They Fail) **Option 1: Increase the limit** ```json PUT /my-index/_settings { "index.mapping.total_fields.limit": 10000 } ``` ::: danger Causes memory issues, slow queries, and cluster instability. ::: **Option 2: Use dynamic mapping** ```json { "mappings": { "dynamic": true } } ``` ::: danger Creates new fields automatically, quickly hitting the limit. ::: **Option 3: Use flattened type** ```json { "custom_data": { "type": "flattened" } } ``` ::: danger Limited query capabilities — no range queries, no aggregations on numeric values. ::: ### The Custom Fields Solution Foundatio.Repositories solves this with **pooled field slots** and **dynamic templates**: ``` Instead of: Use pooled slots: tenant_a_field1 ─┐ idx.string-1 ← All string fields tenant_a_field2 │ idx.string-2 tenant_b_field1 ├─ 1000+ idx.int-1 ← All integer fields tenant_b_field2 │ fields idx.int-2 tenant_c_field1 │ idx.bool-1 ← All boolean fields ... ─┘ idx.date-1 ← All date fields ──────── ~20 fields total ``` **How it works:** 1. Register typed field handlers in your index (e.g., string, int, bool, date) 2. Elasticsearch dynamic templates auto-map `idx.*` sub-fields by type pattern 3. Each tenant's custom field is assigned to an available slot of the matching type 4. Field definitions map logical names to physical slots per tenant 5. Queries are automatically translated from logical names to slot names **Benefits:** * Unlimited logical custom fields across all tenants * Full query and aggregation support * Proper field types (not just strings) * Fixed, predictable mapping size * No risk of mapping explosion ## Setup and Registration ### 1. Implement IHaveCustomFields on Your Entity ```csharp using Foundatio.Repositories.Elasticsearch.CustomFields; using Foundatio.Repositories.Models; public class Employee : IIdentity, IHaveDates, IHaveCustomFields { public string Id { get; set; } = string.Empty; public string Name { get; set; } = string.Empty; public string CompanyId { get; set; } = string.Empty; public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } public IDictionary Data { get; set; } = new Dictionary(); public IDictionary Idx { get; set; } = new Dictionary(); public string GetTenantKey() => CompanyId; } ``` ### Interface Requirements ```csharp public interface IHaveCustomFields : IHaveData { IDictionary Idx { get; } string GetTenantKey(); } public interface IHaveData { IDictionary Data { get; set; } } ``` **Data vs Idx:** * `Data` - Stored but not indexed. Put custom field **values** here. The framework reads from `Data` during save. * `Idx` - Stored and indexed. The framework **automatically populates** this from `Data` during save. Do not set `Idx` values directly. ### 2. Configure Your Index Register custom field types in your index constructor. Call `AddStandardCustomFieldTypes()` to register all built-in types, or register individual types with `AddCustomFieldType()`: ```csharp using Foundatio.Repositories.Elasticsearch.Configuration; using Foundatio.Repositories.Elasticsearch.CustomFields; using Foundatio.Repositories.Elasticsearch.Extensions; public sealed class EmployeeIndex : VersionedIndex { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 1) { AddStandardCustomFieldTypes(); } public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.Id) .Keyword(e => e.CompanyId) .Text(e => e.Name) ); } } ``` ::: tip How It Works Under the Hood * `SetupDefaults()` detects `IHaveCustomFields` on your entity and automatically adds `idx` as a dynamic object field. * When custom field types are registered, Elasticsearch **dynamic templates** are created that auto-map sub-fields under `idx.*` by type pattern (e.g., `string-*`, `int-*`, `bool-*`). * This means you do **not** need to manually define individual slot fields in your mapping. ::: ### 3. Configure Your ElasticConfiguration Call `AddCustomFieldIndex()` in your configuration constructor to create the `CustomFieldDefinition` index and enable the `CustomFieldDefinitionRepository`: ```csharp using Foundatio.Repositories.Elasticsearch.Configuration; using Foundatio.Repositories.Elasticsearch.CustomFields; public class MyAppElasticConfiguration : ElasticConfiguration { public MyAppElasticConfiguration( IQueue workItemQueue, ICacheClient cacheClient, IMessageBus messageBus, ILoggerFactory loggerFactory) : base(workItemQueue, cacheClient, messageBus, loggerFactory: loggerFactory) { AddIndex(Employees = new EmployeeIndex(this)); CustomFields = AddCustomFieldIndex(replicas: 1); } public EmployeeIndex Employees { get; } public CustomFieldDefinitionIndex CustomFields { get; } } ``` `AddCustomFieldIndex()` creates a `CustomFieldDefinitionIndex` (a `VersionedIndex`) and registers it with the configuration. The `CustomFieldDefinitionRepository` is lazily created when first accessed via `configuration.CustomFieldDefinitionRepository`. ### 4. Register DI Services Register the `ICustomFieldDefinitionRepository` singleton by resolving it from your configuration: ```csharp services.AddSingleton(); services.AddSingleton(s => s.GetRequiredService()); services.AddSingleton(s => s.GetRequiredService().CustomFieldDefinitionRepository); ``` ### 5. Configure Your Repository ```csharp using Foundatio.Repositories.Elasticsearch; public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { AutoCreateCustomFields = true; } } ``` When `AutoCreateCustomFields` is `true`, any key in `Data` that doesn't have a matching `CustomFieldDefinition` will automatically get one created as a `string` type. ## Custom Field Definitions ### CustomFieldDefinition Each custom field is tracked by a `CustomFieldDefinition` record stored in a dedicated Elasticsearch index: ```csharp public record CustomFieldDefinition : IIdentity, IHaveDates, ISupportSoftDeletes, IHaveData { public string Id { get; set; } public string EntityType { get; set; } // e.g., "Employee" (immutable after creation) public string TenantKey { get; set; } // Tenant identifier (immutable after creation) public string Name { get; set; } // Friendly field name public string Description { get; set; } // Optional description public int DisplayOrder { get; set; } // UI ordering hint public CustomFieldProcessMode ProcessMode { get; set; } = CustomFieldProcessMode.ProcessOnValue; public int ProcessOrder { get; set; } // Processing sequence within a mode public string IndexType { get; set; } // e.g., "string", "int", "date", "bool" public int IndexSlot { get; set; } // Auto-assigned slot number (immutable after creation) public IDictionary Data { get; set; } = new Dictionary(); public DateTime CreatedUtc { get; set; } public DateTime UpdatedUtc { get; set; } public bool IsDeleted { get; set; } } ``` ::: warning Immutable After Creation `EntityType`, `TenantKey`, and `IndexSlot` cannot be changed after a definition is created. Attempting to modify these via `SaveAsync` throws a `DocumentValidationException`. ::: ### ProcessMode ```csharp public enum CustomFieldProcessMode { ProcessOnValue, // Default: process only when Data contains a value for this field AlwaysProcess // Run the field type processor even when no value is present (for calculated fields) } ``` * `ProcessOnValue` fields are processed first, in `ProcessOrder` order * `AlwaysProcess` fields are processed after all `ProcessOnValue` fields, in `ProcessOrder` order ### ICustomFieldDefinitionRepository The repository provides CRUD operations plus custom field-specific methods: ```csharp public interface ICustomFieldDefinitionRepository : ISearchableRepository { Task> GetFieldMappingAsync( string entityType, string tenantKey); Task> FindByTenantAsync( string entityType, string tenantKey); Task AddFieldAsync( string entityType, string tenantKey, string name, string indexType, string description = null, int displayOrder = 0, IDictionary data = null); } ``` | Method | Description | |--------|-------------| | `GetFieldMappingAsync` | Returns a name-keyed dictionary of all active definitions for a tenant. Cached for 15 minutes. | | `FindByTenantAsync` | Returns paginated results of all definitions for an entity type + tenant (up to 1000 per page). | | `AddFieldAsync` | Convenience method to create a `CustomFieldDefinition` with auto-assigned slot. | The concrete `CustomFieldDefinitionRepository` class also exposes `RemoveByTenantAsync(entityType, tenantKey)` for bulk tenant removal (not on the interface). Since the interface extends `ISearchableRepository`, all standard repository methods are available: `AddAsync`, `SaveAsync`, `RemoveAsync`, `RemoveAllAsync`, `GetByIdAsync`, `GetByIdsAsync`, `FindAsync`, etc. ## Built-in Field Types | Class | IndexType | Elasticsearch Mapping | Slot Pattern | |-------|-----------|----------------------|--------------| | `BooleanFieldType` | `bool` | Boolean | `idx.bool-{slot}` | | `DateFieldType` | `date` | Date | `idx.date-{slot}` | | `DoubleFieldType` | `double` | Number (Double) | `idx.double-{slot}` | | `FloatFieldType` | `float` | Number (Float) | `idx.float-{slot}` | | `IntegerFieldType` | `int` | Number (Integer) | `idx.int-{slot}` | | `KeywordFieldType` | `keyword` | Keyword | `idx.keyword-{slot}` | | `LongFieldType` | `long` | Number (Long) | `idx.long-{slot}` | | `StringFieldType` | `string` | Text + Keyword sub-field | `idx.string-{slot}` | Register all standard types at once with `AddStandardCustomFieldTypes()` in your index constructor, or register individual types with `AddCustomFieldType()` or `AddCustomFieldType(instance)`. ## Using Custom Fields ### Setting Custom Fields Custom field values go in the `Data` dictionary. The framework automatically processes them into `Idx` during save: ```csharp var employee = new Employee { Name = "John Doe", CompanyId = "tenant-123", Data = new Dictionary { ["department"] = "Engineering", ["level"] = 5, ["isRemote"] = true } }; await _repository.AddAsync(employee, o => o.ImmediateConsistency()); ``` ::: warning Do **not** set values directly on `Idx`. The framework clears and repopulates `Idx` from `Data` on every save, using the registered `CustomFieldDefinition` for each field to determine the correct slot. ::: ### Querying Custom Fields Custom fields support automatic field name resolution. Use logical field names in filter expressions — the framework translates them to the correct `idx.*` slot: ```csharp var results = await _repository.FindAsync(q => q .FilterExpression("department:Engineering")); var results = await _repository.FindAsync(q => q .FilterExpression("level:5")); ``` Field name resolution is case-insensitive and requires the query to include a tenant key so the correct field mapping can be loaded. ### Type Mismatches If a value in `Data` does not match the `IndexType` of its `CustomFieldDefinition`, the document will still be saved but Elasticsearch will silently reject the malformed index value. The field will appear to not exist when queried: ```csharp // Definition expects an integer await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "acme", Name = "Level", IndexType = IntegerFieldType.IndexType }); // But we store a string value employee.Data["Level"] = "not-a-number"; await _repository.AddAsync(employee, o => o.ImmediateConsistency()); // This returns NO results — Elasticsearch ignored the malformed value var results = await _repository.FindAsync(q => q .FilterExpression("_exists_:level")); ``` ::: warning The document is saved successfully and `Data["Level"]` will contain `"not-a-number"` when retrieved. However, the value is **not indexed** — it won't appear in search results, `_exists_` checks, or aggregations. Always validate values before saving to avoid silent data loss in the index. ::: ## Custom Field Types ### Implementing ICustomFieldType ```csharp public interface ICustomFieldType { string Type { get; } Task ProcessValueAsync( T document, object value, CustomFieldDefinition fieldDefinition) where T : class; Func, IProperty> ConfigureMapping() where T : class; } public class ProcessFieldValueResult { public object Value { get; set; } public object Idx { get; set; } public bool IsCustomFieldDefinitionModified { get; set; } } ``` | Property | Description | |----------|-------------| | `Value` | The processed value to store back in `Data` | | `Idx` | Optional separate value for the index (if different from `Value`). When `null`, `Value` is used for both. | | `IsCustomFieldDefinitionModified` | Set to `true` if your processor modified the `CustomFieldDefinition` itself (triggers a save). | ### Custom Field Type Example ```csharp public class PercentFieldType : ICustomFieldType { public string Type => "percent"; public Task ProcessValueAsync( T document, object value, CustomFieldDefinition fieldDefinition) where T : class { if (value is int intValue) { var clamped = Math.Clamp(intValue, 0, 100); return Task.FromResult(new ProcessFieldValueResult { Value = clamped }); } return Task.FromResult(new ProcessFieldValueResult { Value = value }); } public Func, IProperty> ConfigureMapping() where T : class { return factory => factory.IntegerNumber(); } } ``` ### Registering Custom Types ```csharp public sealed class EmployeeIndex : VersionedIndex { public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 1) { AddStandardCustomFieldTypes(); AddCustomFieldType(new PercentFieldType()); } } ``` ## Field Definition Lifecycle Management Understanding the full lifecycle of a custom field definition is critical for capacity planning and avoiding slot exhaustion. ### Complete Lifecycle Example This example walks through creating, soft-deleting, reusing names, hard-deleting, and reclaiming slots: ```csharp // Step 1: Create three string fields for tenant "acme" // Slots are assigned sequentially: 1, 2, 3 var field1 = await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "acme", Name = "Department", IndexType = StringFieldType.IndexType }); // field1.IndexSlot == 1, physical field: idx.string-1 var field2 = await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "acme", Name = "Region", IndexType = StringFieldType.IndexType }); // field2.IndexSlot == 2, physical field: idx.string-2 var field3 = await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "acme", Name = "CostCenter", IndexType = StringFieldType.IndexType }); // field3.IndexSlot == 3, physical field: idx.string-3 // Step 2: Soft-delete "Region" — frees the NAME but NOT the slot field2.IsDeleted = true; await _customFieldDefinitionRepository.SaveAsync(field2); var mapping = await _customFieldDefinitionRepository.GetFieldMappingAsync( nameof(Employee), "acme"); // mapping contains "Department" and "CostCenter" — "Region" is excluded // Step 3: Reuse the name "Region" — gets a NEW slot (4), not the old one (2) var field4 = await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "acme", Name = "Region", IndexType = StringFieldType.IndexType }); // field4.IndexSlot == 4 (slot 2 is still occupied by the soft-deleted record) // Step 4: Hard-delete the original soft-deleted "Region" — frees slot 2 await _customFieldDefinitionRepository.RemoveAsync(field2); // Step 5: Next new field gets the freed slot 2 var field5 = await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "acme", Name = "Division", IndexType = StringFieldType.IndexType }); // field5.IndexSlot == 2 (recycled from the hard-deleted record) ``` ::: tip Slot Recycling Summary * **Soft delete** → name freed, slot occupied (allows graceful migration) * **Hard delete** → name freed, slot freed (allows slot reuse) * **To fully free a slot:** soft-delete first, then hard-delete once you're confident existing data has been migrated or is no longer needed ::: ### Creating Definitions Create definitions explicitly via `AddAsync` or the `AddFieldAsync` convenience method: ```csharp var definition = await _customFieldDefinitionRepository.AddAsync(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "tenant-123", Name = "department", IndexType = StringFieldType.IndexType }); ``` Or use the convenience method: ```csharp var definition = await _customFieldDefinitionRepository.AddFieldAsync( entityType: nameof(Employee), tenantKey: "tenant-123", name: "department", indexType: "string", description: "Employee department"); ``` **Slot assignment** is automatic. Slots start at 1 and increment per `(EntityType, TenantKey, IndexType)` scope. You cannot pre-assign a slot — `IndexSlot` must be 0 when calling `AddAsync`. **Duplicate handling:** * Adding a field with the **same name and same type** as an existing **active** definition silently returns the existing definition. * Adding a field with the **same name but different type** throws `DocumentValidationException`. * Adding a field with the **same name** as a **soft-deleted** definition creates a **new** definition with a **new slot**. The soft-deleted record's slot remains occupied until it is hard-deleted. ### Updating Definitions Use `SaveAsync` to update mutable properties: ```csharp definition.Description = "Updated description"; definition.DisplayOrder = 5; await _customFieldDefinitionRepository.SaveAsync(definition); ``` **Mutable properties:** `Name`, `Description`, `DisplayOrder`, `ProcessMode`, `ProcessOrder`, `Data`, `IsDeleted` **Immutable properties (enforced at save time):** `EntityType`, `TenantKey`, `IndexSlot` ### Soft Deleting Definitions Soft delete frees the field **name** for reuse but the **slot remains occupied**: ```csharp definition.IsDeleted = true; await _customFieldDefinitionRepository.SaveAsync(definition); ``` After soft deletion: * The field name can be reused by a new definition (assigned a new slot) * The field is excluded from `GetFieldMappingAsync` results * The slot is **not** freed — it cannot be reused until the definition is hard-deleted * The definition is still queryable with `IncludeSoftDeletes()` option ### Hard Deleting Definitions Hard delete frees **both** the name and the slot for reuse: ```csharp await _customFieldDefinitionRepository.RemoveAsync(definition); ``` After hard deletion, the freed slot number will be reassigned to the next new field of the same type for that tenant. ### Bulk Operations ```csharp // Find all definitions for a tenant var tenantFields = await _customFieldDefinitionRepository.FindByTenantAsync( nameof(Employee), "tenant-123"); var allFields = new List(); do { allFields.AddRange(tenantFields.Documents); } while (await tenantFields.NextPageAsync()); // Hard-delete all definitions for a tenant (frees all slots) await _customFieldDefinitionRepository.RemoveAllAsync(q => q .FieldEquals(d => d.EntityType, nameof(Employee)) .FieldEquals(d => d.TenantKey, "tenant-123")); ``` ## Cleanup Patterns ::: warning There is **no built-in cleanup job** for custom field definitions. Applications must manage the lifecycle of their custom field definitions. ::: ### Tenant Offboarding When a tenant is removed, hard-delete all their custom field definitions to free slots: ```csharp await _customFieldDefinitionRepository.RemoveAllAsync(q => q .FieldEquals(d => d.EntityType, nameof(Employee)) .FieldEquals(d => d.TenantKey, "tenant-123")); ``` ### Slot Reclamation Soft-deleted definitions still occupy slots. Periodically hard-delete old soft-deleted definitions to reclaim them: ```csharp await _customFieldDefinitionRepository.RemoveAllAsync(q => q .FieldEquals(d => d.IsDeleted, true) .FieldEquals(d => d.EntityType, nameof(Employee)) .DateRange(null, DateTime.UtcNow.AddDays(-30), (CustomFieldDefinition d) => d.UpdatedUtc), o => o.IncludeSoftDeletes()); ``` ### Synchronizing With Domain Model Changes When your domain model controls which custom fields exist (e.g., tenant settings define available fields), you need to keep `CustomFieldDefinition` records in sync. A common pattern is subscribing to `DocumentsChanged` events and comparing the original vs. modified documents. Below is a **simplified** example. Real-world implementations will handle more edge cases depending on your domain model. ```csharp public class CustomFieldSyncService : IStartupAction { private readonly ICustomFieldDefinitionRepository _customFieldDefinitionRepository; private readonly ITenantSettingsRepository _settingsRepository; public CustomFieldSyncService( ICustomFieldDefinitionRepository customFieldDefinitionRepository, ITenantSettingsRepository settingsRepository) { _customFieldDefinitionRepository = customFieldDefinitionRepository; _settingsRepository = settingsRepository; } public Task RunAsync(CancellationToken shutdownToken = default) { _settingsRepository.DocumentsChanged.AddHandler((_, args) => SynchronizeCustomFieldsAsync(args.Documents)); return Task.CompletedTask; } private async Task SynchronizeCustomFieldsAsync( IReadOnlyCollection> changes) { var toAdd = new List(); var toDelete = new List(); foreach (var change in changes) { string tenantKey = change.Value?.Id ?? change.Original.Id; var originalFieldNames = (change.Original?.FieldNames ?? []).ToHashSet(); var modifiedFieldNames = (change.Value?.FieldNames ?? []).ToHashSet(); // New fields: in modified but not in original foreach (string name in modifiedFieldNames.Except(originalFieldNames)) { toAdd.Add(new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = tenantKey, Name = name, IndexType = StringFieldType.IndexType }); } // Removed fields: in original but not in modified — soft-delete them var removedNames = originalFieldNames.Except(modifiedFieldNames).ToHashSet(); if (removedNames.Count > 0) { var existing = await _customFieldDefinitionRepository.FindByTenantAsync( nameof(Employee), tenantKey); do { toDelete.AddRange(existing.Documents.Where(f => removedNames.Contains(f.Name))); } while (await existing.NextPageAsync()); } } foreach (var def in toDelete) def.IsDeleted = true; if (toDelete.Count > 0) await _customFieldDefinitionRepository.SaveAsync(toDelete); if (toAdd.Count > 0) await _customFieldDefinitionRepository.AddAsync(toAdd); } } ``` ::: tip Soft-delete before adding. When soft-deleting and adding fields in the same batch, process deletes first. This ensures that any freed names are available for the new definitions and avoids name collision errors from orphaned records. ::: ## Slot Management Custom fields use pooled slots in the index mapping to avoid mapping explosions. ### How Slots Work ``` Logical Field Name Slot Assignment Physical Field ───────────────── ──────────────── ────────────── Tenant A: "department" → string slot 1 → idx.string-1 Tenant A: "region" → string slot 2 → idx.string-2 Tenant B: "department" → string slot 1 → idx.string-1 (same slot, different tenant) Tenant B: "priority" → int slot 1 → idx.int-1 Tenant C: "projectId" → string slot 1 → idx.string-1 ``` Each tenant gets their own slot namespace per `(EntityType, TenantKey, IndexType)`, so "department" for Tenant A and "department" for Tenant B both map to `idx.string-1` but are isolated by tenant-scoped queries. ### Slot Naming Convention Slot names follow the pattern `{IndexType}-{IndexSlot}`: ``` idx.string-1, idx.string-2, idx.string-3... - String/text slots idx.keyword-1, idx.keyword-2... - Keyword slots idx.int-1, idx.int-2... - Integer slots idx.double-1, idx.double-2... - Double slots idx.bool-1, idx.bool-2... - Boolean slots idx.date-1, idx.date-2... - Date slots ``` These are **automatically mapped** by Elasticsearch dynamic templates. You do not need to pre-declare individual slot fields. ### Slot Exhaustion Elasticsearch dynamic templates can create new sub-fields on demand, so slot capacity is effectively unlimited per type. However, each additional slot increases the total field count in the index. Monitor your total field usage relative to Elasticsearch's `index.mapping.total_fields.limit`. ## Calculated / Computed Fields Use `ProcessMode = CustomFieldProcessMode.AlwaysProcess` to create fields that are computed from other field values during save. Combined with a custom `ICustomFieldType`, this enables derived fields. ### Processing Order 1. **`ProcessOnValue` fields** run first — only when a matching key exists in `Data` 2. **`AlwaysProcess` fields** run after all `ProcessOnValue` fields — regardless of whether a value exists 3. Within each phase, fields are processed in `ProcessOrder` order ### Example: Calculated Integer Field Define the custom field type: ```csharp public class CalculatedIntegerFieldType : IntegerFieldType { private readonly ScriptService _scriptService; public CalculatedIntegerFieldType(ScriptService scriptService) { _scriptService = scriptService; } public override async Task ProcessValueAsync( T document, object value, CustomFieldDefinition fieldDefinition) where T : class { if (!fieldDefinition.Data.TryGetValue("Expression", out object expression)) return await base.ProcessValueAsync(document, value, fieldDefinition); var result = await _scriptService.EvaluateForSourceAsync(document, expression.ToString()); if (result.IsCancelled || result.Value is Double.NaN) return new ProcessFieldValueResult { Value = null }; return new ProcessFieldValueResult { Value = result.Value }; } } ``` Register it in your index: ```csharp AddStandardCustomFieldTypes(); AddCustomFieldType(new CalculatedIntegerFieldType(scriptService)); ``` Create the calculated field definition with an expression in `Data`: ```csharp await _customFieldDefinitionRepository.AddAsync([ new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "1", Name = "Field1", IndexType = IntegerFieldType.IndexType }, new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "1", Name = "Field2", IndexType = IntegerFieldType.IndexType }, new CustomFieldDefinition { EntityType = nameof(Employee), TenantKey = "1", Name = "Calculated", IndexType = IntegerFieldType.IndexType, ProcessMode = CustomFieldProcessMode.AlwaysProcess, Data = new Dictionary { { "Expression", "source.Data.Field1 + source.Data.Field2" } } } ]); ``` Now when a document is saved with `Field1 = 1` and `Field2 = 2`, the `Calculated` field automatically computes to `3` and is indexed for querying: ```csharp employee.Data["Field1"] = 1; employee.Data["Field2"] = 2; await _repository.AddAsync(employee, o => o.ImmediateConsistency()); var results = await _repository.FindAsync(q => q .FilterExpression("calculated:3")); ``` ## Virtual Custom Fields For entities where custom fields are not stored in a flat `Data` dictionary, implement `IHaveVirtualCustomFields` instead of `IHaveCustomFields`: ```csharp public interface IHaveVirtualCustomFields { IDictionary GetCustomFields(); object GetCustomField(string name); void SetCustomField(string name, object value); void RemoveCustomField(string name); IDictionary Idx { get; } string GetTenantKey(); } ``` This gives you full control over how custom field values are read and written, while the framework still handles slot assignment, `Idx` population, and query field resolution. ## Concurrency and Locking ### Thread-Safe Slot Allocation Slot allocation uses **distributed locks** (via `CacheLockProvider`) scoped per `(EntityType, TenantKey, IndexType)` to prevent duplicate slot assignment under concurrent writes. The lock key follows the pattern `customfield:{entityType}:{tenantKey}:{indexType}`. ### Caching Behavior | Cache | TTL | Key Pattern | Description | |-------|-----|-------------|-------------| | Field mapping | 15 min | `customfield:{entityType}:{tenantKey}` | Name-to-definition dictionary | | Available slots | 5 min | `customfield:{entityType}:{tenantKey}:{indexType}:slots` | List of free slot numbers | | Used names | 5 min | `customfield:{entityType}:{tenantKey}:names` | Set of active field names | Cache invalidation happens automatically on add, save, and remove operations. Bulk removal (`RemoveAllAsync`) clears all custom field caches by prefix when the affected scope cannot be determined from the query. ### Consistency The `CustomFieldDefinitionRepository` defaults to `Consistency.Immediate` (all writes use `refresh=wait_for`), ensuring that newly created definitions are immediately visible for subsequent queries and slot allocation checks. ## Best Practices ### 1. Use Appropriate Field Types ```csharp // Good: use correct types for proper indexing and querying employee.Data["count"] = 42; // int employee.Data["price"] = 19.99; // double employee.Data["isActive"] = true; // bool employee.Data["createdAt"] = DateTime.UtcNow; // date // Bad: storing everything as strings loses type-specific query capabilities employee.Data["count"] = "42"; ``` ### 2. Index Only What You Query ```csharp // Queryable data goes in Data (gets indexed via CustomFieldDefinitions) employee.Data["searchableField"] = "value"; // Large or rarely queried data can also go in Data without a definition // (it will be stored but not indexed if there's no matching definition // and AutoCreateCustomFields is false) ``` ### 3. Handle Missing Fields ```csharp if (employee.Data.TryGetValue("department", out var dept)) { Console.WriteLine($"Department: {dept}"); } ``` ### 4. Plan for Cleanup * Always soft-delete before hard-deleting to allow graceful migration * Implement periodic cleanup of soft-deleted definitions older than a threshold * Clean up definitions when tenants are offboarded * Monitor slot usage relative to Elasticsearch field limits ### 5. Design Tenant Keys Carefully The `TenantKey` returned by `GetTenantKey()` scopes all custom field definitions. Choose a key that matches your multi-tenancy boundary: ```csharp // Simple: one set of custom fields per company public string GetTenantKey() => CompanyId; // Composite: separate custom fields per company + entity subtype public string GetTenantKey() => $"{CompanyId}-{SubType}"; ``` ::: tip Tenant Key Guidelines * Keep tenant keys **as simple as possible** — use only the fields that define your tenancy boundary. * Each unique tenant key gets its own independent pool of field slots and names. * More granular keys mean more isolation but also more `CustomFieldDefinition` records to manage. * Tenant keys are **immutable** on `CustomFieldDefinition` — plan your key structure before deploying. ::: ## Next Steps * [Querying](/guide/querying) - Query custom fields * [Index Management](/guide/index-management) - Configure index mappings * [Configuration](/guide/configuration) - Custom field configuration --- --- url: /guide/troubleshooting.md --- # Troubleshooting This guide covers common issues and solutions when working with Foundatio.Repositories. ## Connection Issues ### Cannot Connect to Elasticsearch **Symptoms:** * `No connection could be made` * `Connection refused` * Timeout errors **Solutions:** 1. **Verify Elasticsearch is running:** ```bash curl http://localhost:9200 ``` 2. **Check connection string:** ```csharp protected override NodePool CreateConnectionPool() { // Ensure URL is correct return new SingleNodePool(new Uri("http://localhost:9200")); } ``` 3. **Check firewall/network:** ```bash # Test connectivity telnet localhost 9200 ``` 4. **Enable debug logging:** ```csharp protected override void ConfigureSettings(ElasticsearchClientSettings settings) { settings.DisableDirectStreaming(); settings.PrettyJson(); } ``` ### Authentication Errors **Symptoms:** * `401 Unauthorized` * `403 Forbidden` **Solutions:** ```csharp protected override void ConfigureSettings(ElasticsearchClientSettings settings) { // Basic authentication settings.Authentication(new BasicAuthentication("username", "password")); // Or API key settings.Authentication(new ApiKey("encoded-api-key")); } ``` ## Index Issues ### Index Not Found **Symptoms:** * `index_not_found_exception` * `no such index` **Solutions:** 1. **Configure indexes on startup:** ```csharp await configuration.ConfigureIndexesAsync(); ``` 2. **Check index name:** ```csharp // Versioned indexes have version suffix // "employees" -> "employees-v1" var indexName = index.VersionedName; ``` 3. **Verify index exists:** ```bash curl http://localhost:9200/_cat/indices ``` ### Mapping Conflicts > See [Mapping Lifecycle](/guide/index-management#mapping-lifecycle) for a complete breakdown of how and when mappings are applied per index type, including important differences for `DailyIndex`/`MonthlyIndex`. **Symptoms:** * `mapper_parsing_exception` * `failed to parse field` **Solutions:** 1. **Increment index version:** ```csharp // Change version to trigger reindex public EmployeeIndex(...) : base(configuration, "employees", version: 2) { } ``` 2. **Delete and recreate index (development only):** ```csharp await configuration.DeleteIndexesAsync(); await configuration.ConfigureIndexesAsync(); ``` 3. **Check field types match:** ```csharp // Ensure mapping matches data types .IntegerNumber(e => e.Age) ``` ## Query Issues ### No Results Returned **Symptoms:** * Empty results when data exists * `Total: 0` **Solutions:** 1. **Check soft delete mode:** ```csharp // Include soft-deleted documents var results = await repository.FindAsync(query, o => o.IncludeSoftDeletes()); ``` 2. **Use immediate consistency:** ```csharp // Wait for index refresh await repository.AddAsync(entity, o => o.ImmediateConsistency()); var results = await repository.FindAsync(query); ``` 3. **Verify filter syntax:** ```csharp // Check filter expression var results = await repository.FindAsync(q => q.FieldEquals(e => e.Status, "active")); // Debug: Log the query var results = await repository.FindAsync(query, o => o.QueryLogLevel(LogLevel.Debug)); ``` 4. **Check field names:** ```csharp // Use exact field names from mapping // "name" vs "name.keyword" for exact match ``` ### Query Syntax Errors **Symptoms:** * `query_parsing_exception` * `failed to parse query` **Solutions:** 1. **Escape special characters:** ```csharp // Escape: + - = && || > < ! ( ) { } [ ] ^ " ~ * ? : \ / var escaped = Regex.Escape(userInput); ``` 2. **Use strongly-typed queries:** ```csharp // Instead of filter expression var results = await repository.FindAsync(q => q .FieldEquals(e => e.Status, "active") .FieldCondition(e => e.Name, ComparisonOperator.Contains, "John")); ``` For numeric comparisons, use `FilterExpression` with Lucene syntax: ```csharp var results = await repository.FindAsync(q => q .FieldEquals(e => e.Status, "active") .FilterExpression("age:[25 TO *]")); ``` ## Cache Issues ### Stale Data **Symptoms:** * Old data returned after updates * Changes not reflected **Solutions:** 1. **Manually invalidate cache:** ```csharp await repository.InvalidateCacheAsync(document); await repository.InvalidateCacheAsync("custom-cache-key"); ``` 2. **Disable cache for debugging:** ```csharp var results = await repository.FindAsync(query, o => o.Cache(false)); ``` 3. **Check cache invalidation gaps:** ```csharp // PatchAllAsync doesn't invalidate custom keys await repository.PatchAllAsync(query, patch); await repository.InvalidateCacheAsync("affected-key"); ``` ### Cache Key Conflicts **Symptoms:** * Wrong data returned * Data from different queries mixed **Solutions:** ```csharp // Use unique, consistent cache keys var key = $"employee:email:{email.ToLowerInvariant()}"; var results = await repository.FindOneAsync(query, o => o.Cache(key)); ``` ## Version Conflicts ### VersionConflictDocumentException **Symptoms:** * `version_conflict_engine_exception` * `VersionConflictDocumentException` **Solutions:** 1. **Implement retry logic:** ```csharp int retries = 3; while (retries > 0) { try { var doc = await repository.GetByIdAsync(id); doc.Name = "Updated"; await repository.SaveAsync(doc); break; } catch (VersionConflictDocumentException) { retries--; if (retries == 0) throw; } } ``` 2. **Skip version check (if appropriate):** ```csharp await repository.SaveAsync(document, o => o.SkipVersionCheck()); ``` 3. **Use atomic operations:** ```csharp // Atomic increment avoids conflicts await repository.PatchAsync(id, new ScriptPatch("ctx._source.counter++")); ``` ## Performance Issues ### Slow Queries **Symptoms:** * High query latency * Timeouts **Solutions:** 1. **Add appropriate indexes:** ```csharp // Ensure fields are properly mapped .Keyword(f => f.Name(e => e.Status)) // For filtering .Text(f => f.Name(e => e.Name).AddKeywordAndSortFields()) // For search + sort ``` 2. **Limit result size:** ```csharp var results = await repository.FindAsync(query, o => o.PageLimit(100)); ``` 3. **Use field selection:** ```csharp var results = await repository.FindAsync(query, o => o .Include(e => e.Id) .Include(e => e.Name)); ``` 4. **Use search-after for deep pagination:** ```csharp var results = await repository.FindAsync(query, o => o.SearchAfterPaging()); ``` ### Memory Issues **Symptoms:** * `OutOfMemoryException` * High memory usage **Solutions:** 1. **Use batch processing:** ```csharp await repository.BatchProcessAsync(query, async batch => { // Process in batches return true; }, o => o.PageLimit(500)); ``` 2. **Use snapshot paging for large exports:** ```csharp var results = await repository.FindAsync(query, o => o.SnapshotPaging()); ``` ### Reindex Rejected Due to Indexing Pressure **Symptoms:** * A reindex fails (or the reindex task status can no longer be retrieved) with a server error like: ```text Server Error (Index=): rejected execution of coordinating operation [coordinating_and_primary_bytes=0, replica_bytes=0, all_bytes=0, coordinating_operation_bytes=158478331, max_coordinating_bytes=107374182] ``` * The error `type` is `es_rejected_execution_exception` with an HTTP `429` status. * Repeated `Error getting task status while reindexing: "{OldIndex}" -> "{NewIndex}"` log entries, possibly followed by `Failed to get the status {N} times in a row for reindex task ...`. **Cause:** Elasticsearch reserves a portion of JVM heap for in-flight indexing work — the [`indexing_pressure.memory.limit`](https://www.elastic.co/docs/reference/elasticsearch/configuration-reference/indexing-pressure-settings) node setting, which defaults to **10% of heap**. Every indexing request (including the internal bulk writes issued by the `_reindex` API) is accounted against this budget for the full duration of its coordinating/primary/replica stages. If a single bulk sub-request's estimated size exceeds the remaining budget, Elasticsearch immediately rejects it rather than queuing it — this is a deliberate back-pressure mechanism, not a bug. It's more likely to trigger during reindexing because reindex batches (default 1000 documents) scale with document size: large documents can produce a bulk payload well over 100MB on a modestly sized node. See [Rejected requests: Analyze indexing pressure](https://www.elastic.co/docs/troubleshoot/elasticsearch/rejected-requests#analyze-indexing-pressure) for the full explanation. **Solutions:** 1. **Reduce the reindex batch size** so each internal bulk sub-request stays well under the indexing pressure limit: ```csharp public EmployeeIndex(IElasticConfiguration configuration) : base(configuration, "employees", version: 2) { ReindexBatchSize = 200; // default: 1000 } ``` 2. **Throttle the reindex** to reduce sustained load on a cluster that's also serving other traffic: ```csharp ReindexRequestsPerSecond = 500; // default: unlimited ``` 3. **Check node heap and `indexing_pressure.memory.limit`** via the [node stats API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-nodes-stats) if rejections continue after lowering the batch size — the node may simply be undersized for the document volume/size involved. See [Throttling Reindex Load](./index-management.md#throttling-reindex-load) for more on both properties. Reindex task-status polling also backs off automatically (starting at 1 second, doubling up to a 30 second cap, with +/-25% jitter) after this library's fix for this exact scenario, so transient rejections while polling no longer retry in a tight loop or in lockstep with other reindex operations hitting the same cluster-wide condition. If you configure a low `ReindexRequestsPerSecond` to work around this, note that the reindex's stall-detection timeout (10 minutes by default) automatically extends to accommodate the resulting longer pause between batches, so throttling to avoid indexing pressure rejections won't itself cause the reindex to be cancelled as falsely "stalled." ## Notification Issues ### EntityChanged Not Received **Symptoms:** * Subscribers not receiving notifications * Message bus appears silent **Solutions:** 1. **Verify message bus is configured:** ```csharp public MyElasticConfiguration(IMessageBus messageBus, ...) : base(messageBus: messageBus, ...) { } ``` 2. **Check notifications are enabled:** ```csharp // Repository level NotificationsEnabled = true; // Operation level await repository.SaveAsync(entity, o => o.Notifications(true)); ``` 3. **Verify subscription:** ```csharp await messageBus.SubscribeAsync(async (msg, ct) => { Console.WriteLine($"Received: {msg.Type} {msg.ChangeType}"); }); ``` ### Soft Delete Not Sending Removed **Symptoms:** * Soft delete sends `ChangeType.Saved` instead of `Removed` **Solutions:** ```csharp // Enable originals tracking public class EmployeeRepository : ElasticRepositoryBase { public EmployeeRepository(EmployeeIndex index) : base(index) { OriginalsEnabled = true; // Required for soft delete detection } } ``` ## Debugging Tips ### Enable Detailed Logging ```csharp // In configuration protected override void ConfigureSettings(ElasticsearchClientSettings settings) { settings.DisableDirectStreaming(); settings.PrettyJson(); } // Per query var results = await repository.FindAsync(query, o => o.QueryLogLevel(LogLevel.Debug)); ``` ### Inspect Elasticsearch Directly ```bash # Check cluster health curl http://localhost:9200/_cluster/health # List indexes curl http://localhost:9200/_cat/indices # View mapping curl http://localhost:9200/employees/_mapping # Search directly curl -X POST http://localhost:9200/employees/_search -H 'Content-Type: application/json' -d ' { "query": { "match_all": {} } }' ``` ### Check Index Statistics ```bash curl http://localhost:9200/employees/_stats ``` ## Common Error Messages | Error | Cause | Solution | |-------|-------|----------| | `index_not_found_exception` | Index doesn't exist | Run `ConfigureIndexesAsync()` | | `mapper_parsing_exception` | Type mismatch | Check field types in mapping | | `version_conflict_engine_exception` | Concurrent modification | Implement retry or skip version check | | `search_phase_execution_exception` | Query error | Check query syntax | | `circuit_breaking_exception` | Memory limit | Reduce batch size | | `cluster_block_exception` | Cluster read-only | Check disk space | | `es_rejected_execution_exception` ("rejected execution of coordinating operation") | Indexing pressure limit exceeded, often during reindex of large documents | Lower `ReindexBatchSize`/`ReindexRequestsPerSecond`, see [Reindex Rejected Due to Indexing Pressure](#reindex-rejected-due-to-indexing-pressure) | ## Repository Exception Types Foundatio.Repositories uses typed exceptions so callers can handle specific failure modes. The exceptions listed below inherit from `DocumentException`. | Exception | When Thrown | Retryable? | |-----------|------------|------------| | `DuplicateDocumentException` | `AddAsync` when a document with the same ID already exists | No — remove the existing document or use `SaveAsync` | | `VersionConflictDocumentException` | `SaveAsync` / `PatchAsync` when the document version doesn't match (HTTP 409) | Yes — re-fetch the document and retry | | `DocumentNotFoundException` | `PatchAsync` when the target document doesn't exist (HTTP 404) | No — verify the document ID | | `DocumentValidationException` | Any write operation when document validation fails | No — fix the document data | | `DocumentException` | Other Elasticsearch errors not covered above | Depends on the underlying cause | ### Partial Failures on Bulk Operations When `AddAsync` or `SaveAsync` is called with multiple documents, some may succeed and others may fail. The repository: 1. **Processes all successes first** — fires events, populates cache, sends notifications. 2. **Leaves failed documents' cache unchanged** — failed writes don't mutate Elasticsearch, so existing cache entries remain valid. The writer that caused a conflict handles its own cache update via message bus notifications. 3. **Throws a typed exception** — `DuplicateDocumentException` for add, `VersionConflictDocumentException` for save. ```csharp try { await repository.AddAsync(documents); } catch (DuplicateDocumentException ex) { // Successful documents were fully processed. // Duplicate documents: existing cache entries preserved (nothing was mutated). _logger.LogWarning(ex, "Partial failure: some documents already existed"); } catch (VersionConflictDocumentException ex) { _logger.LogWarning(ex, "Partial failure: some documents had version conflicts"); } ``` ### Transient Error Retries The repository automatically retries transient Elasticsearch errors: * **HTTP 429** (Too Many Requests) — retried with exponential backoff, up to 3 retries (4 total attempts) * **HTTP 503** (Service Unavailable) — retried with exponential backoff, up to 3 retries (4 total attempts) * **HTTP 409** (Version Conflict) — **not** retried; the caller must handle conflict resolution * `DuplicateDocumentException` — **not** retried by the resilience policy ::: info Reindex task-status polling uses its own backoff This resilience policy covers the initial reindex kickoff request. Once reindexing has started, progress is monitored by repeatedly polling the Elasticsearch task status API, which is a plain "did this succeed" response rather than a thrown exception — so it isn't covered by the policy above. That polling loop has its own dedicated exponential backoff (1 second, doubling up to a 30 second cap, with +/-25% jitter so concurrent reindex operations failing for the same reason don't retry in lockstep) on failure. See [Reindex Rejected Due to Indexing Pressure](#reindex-rejected-due-to-indexing-pressure) for the scenario this protects against. ::: ## Aggregation Warnings ### doc\_count\_error\_upper\_bound Warning **Symptoms:** * Warning-level log message about `doc_count_error_upper_bound` in terms aggregation results **Explanation:** When running terms aggregations across multiple shards, Elasticsearch returns an approximate count. The `doc_count_error_upper_bound` field indicates the maximum potential error in document counts for each term bucket. A non-zero value means shard-level approximations may have affected the results. **Solutions:** 1. **Increase `shard_size`** if accuracy matters for your use case — this makes Elasticsearch consider more terms per shard before combining results 2. **Use a single shard** for small indexes where exact counts are important 3. **Ignore the warning** if approximate counts are acceptable for your use case (this is common for analytics and dashboards) ## Getting Help 1. **Check logs** - Enable debug logging 2. **Inspect Elasticsearch** - Use Kibana or curl 3. **Review documentation** - Check specific feature guides 4. **GitHub Issues** - Search existing issues or create new one 5. **Discord** - Join the Foundatio Discord community ## Next Steps * [Configuration](/guide/configuration) - Configuration options * [Elasticsearch Setup](/guide/elasticsearch-setup) - Connection setup * [Caching](/guide/caching) - Cache troubleshooting --- --- url: /guide/upgrading-to-elastic-clients-elasticsearch.md --- # Migrating to Elastic.Clients.Elasticsearch This guide covers breaking changes when upgrading from `NEST` (ES7) to `Elastic.Clients.Elasticsearch` (ES8/ES9). The new Elasticsearch .NET client is a complete rewrite with a new API surface, so most code that interacts with Elasticsearch directly will need changes. > **Query syntax changes**: If you use [Foundatio.Parsers](https://github.com/FoundatioFx/Foundatio.Parsers) for query parsing (e.g., `ElasticQueryParser`, `ElasticMappingResolver`, aggregation parsing), refer to the [Foundatio.Parsers documentation](https://github.com/FoundatioFx/Foundatio.Parsers) for migration notes aligned with `Elastic.Clients.Elasticsearch`. The query parser APIs have been updated to work with the new client types. ## Package Changes **Before:** ```xml ``` **After:** ```xml ``` ## Namespace Changes Remove old NEST namespaces and add new ones: ```csharp // Remove: using Elasticsearch.Net; using Nest; // Add: using Elastic.Clients.Elasticsearch; using Elastic.Clients.Elasticsearch.Mapping; using Elastic.Clients.Elasticsearch.IndexManagement; using Elastic.Transport; ``` Additional namespaces you may need depending on usage: | Feature | Namespace | |---------|-----------| | Aggregations | `Elastic.Clients.Elasticsearch.Aggregations` | | Bulk operations | `Elastic.Clients.Elasticsearch.Core.Bulk` | | Search types | `Elastic.Clients.Elasticsearch.Core.Search` | | Async search | `Elastic.Clients.Elasticsearch.AsyncSearch` | | Analysis (analyzers, tokenizers) | `Elastic.Clients.Elasticsearch.Analysis` | | Fluent helpers | `Elastic.Clients.Elasticsearch.Fluent` | ## ElasticConfiguration Changes ### Client Type | Before | After | |--------|-------| | `IElasticClient Client` | `ElasticsearchClient Client` | | `new ElasticClient(settings)` | `new ElasticsearchClient(settings)` | ### Connection Pool | Before | After | |--------|-------| | `IConnectionPool` | `NodePool` | | `new SingleNodeConnectionPool(uri)` | `new SingleNodePool(uri)` | | `new StaticConnectionPool(nodes)` | `new StaticNodePool(nodes)` | | `new SniffingConnectionPool(nodes)` | `new SniffingNodePool(nodes)` | ### Settings | Before | After | |--------|-------| | `ConnectionSettings` | `ElasticsearchClientSettings` | | `settings.BasicAuthentication(u, p)` | `settings.Authentication(new BasicAuthentication(u, p))` | | `settings.ApiKeyAuthentication(id, key)` | `settings.Authentication(new ApiKey(encoded))` | **Before:** ```csharp protected override IConnectionPool CreateConnectionPool() { return new SingleNodeConnectionPool(new Uri("http://localhost:9200")); } protected override void ConfigureSettings(ConnectionSettings settings) { base.ConfigureSettings(settings); settings.BasicAuthentication("user", "pass"); } ``` **After:** ```csharp protected override NodePool CreateConnectionPool() { return new SingleNodePool(new Uri("http://localhost:9200")); } protected override void ConfigureSettings(ElasticsearchClientSettings settings) { base.ConfigureSettings(settings); settings.Authentication(new BasicAuthentication("user", "pass")); } ``` ### Constructor: Serializer Parameter `ElasticConfiguration` now accepts an `ITextSerializer` parameter. If you don't provide one, a default `SystemTextJsonSerializer` is created with `ConfigureFoundatioRepositoryDefaults()`. If you have custom serialization needs, pass your own serializer: ```csharp var serializer = new SystemTextJsonSerializer( new JsonSerializerOptions().ConfigureFoundatioRepositoryDefaults()); var config = new MyElasticConfiguration( serializer: serializer, cacheClient: cache, messageBus: bus); ``` ### Client Disposal `ElasticsearchClientSettings` implements `IDisposable` internally but doesn't expose it on its public API. The `ElasticConfiguration.Dispose()` method now handles this by casting to `IDisposable`. If you manage the client lifecycle yourself, be aware of this. ## Serialization Changes (Newtonsoft.Json to System.Text.Json) This is one of the largest breaking changes. The new Elasticsearch client uses **System.Text.Json** instead of Newtonsoft.Json for all serialization. ### What Changed | Before | After | |--------|-------| | `NEST.JsonNetSerializer` package | **Removed** — no longer needed or supported | | `Newtonsoft.Json.JsonConverter` | `System.Text.Json.Serialization.JsonConverter` | | `[JsonProperty("name")]` | `[JsonPropertyName("name")]` | | `[JsonIgnore]` (Newtonsoft) | `[JsonIgnore]` (System.Text.Json — same name, different namespace) | | `[JsonConverter(typeof(...))]` (Newtonsoft) | `[JsonConverter(typeof(...))]` (System.Text.Json) | | `JsonConvert.SerializeObject(obj)` | `JsonSerializer.Serialize(obj, options)` | | `JsonConvert.DeserializeObject(json)` | `JsonSerializer.Deserialize(json, options)` | ### ConfigureFoundatioRepositoryDefaults Foundatio.Repositories provides a `ConfigureFoundatioRepositoryDefaults()` extension method on `JsonSerializerOptions` that registers converters needed for correct round-tripping of repository documents: ```csharp using Foundatio.Repositories.Serialization; var options = new JsonSerializerOptions().ConfigureFoundatioRepositoryDefaults(); ``` This registers: * `DoubleSystemTextJsonConverter` to preserve decimal points on whole-number doubles * `ObjectToInferredTypesConverter` to deserialize `object`-typed properties as CLR primitives instead of `JsonElement` (required for `Dictionary` metadata bags unless you supply a custom dictionary converter) * Case-insensitive property matching System.Text.Json serializes enums as **integers** by default, same as Newtonsoft.Json/NEST unless you opted in with `[JsonConverter(typeof(StringEnumConverter))]` or similar. No change is required for typical repository documents. Only add `[JsonConverter(typeof(JsonStringEnumConverter))]` (or a custom converter) on enums you intentionally store as strings in Elasticsearch `_source`. ### LazyDocument Serializer Requirement `LazyDocument` no longer falls back to a default Newtonsoft serializer. The `ITextSerializer` parameter is now **required**: **Before:** ```csharp new LazyDocument(data, serializer: null); // fell back to JsonNetSerializer ``` **After:** ```csharp new LazyDocument(data, serializer); // serializer is required, throws if null ``` ### Migration Tips for Custom Converters If you have custom Newtonsoft `JsonConverter` implementations: 1. Create a new class inheriting from `System.Text.Json.Serialization.JsonConverter` 2. Implement `Read` and `Write` methods using `Utf8JsonReader`/`Utf8JsonWriter` 3. Register converters via `JsonSerializerOptions.Converters.Add(...)` or the `[JsonConverter]` attribute 4. Be aware that System.Text.Json is stricter by default (no comments, trailing commas, or unquoted property names) ## Index Configuration Changes ### ConfigureIndex: Return Type and Descriptor `ConfigureIndex` changed from returning a descriptor (fluent chaining) to `void` (mutating the descriptor in place). The descriptor type also changed: | Before | After | |--------|-------| | `CreateIndexDescriptor ConfigureIndex(CreateIndexDescriptor idx)` | `void ConfigureIndex(CreateIndexRequestDescriptor idx)` | | Returns the descriptor | Mutates the descriptor in place | **Before:** ```csharp public override CreateIndexDescriptor ConfigureIndex(CreateIndexDescriptor idx) { return base.ConfigureIndex(idx .Settings(s => s.NumberOfReplicas(0)) .Map(m => m.AutoMap().Properties(p => p.SetupDefaults()))); } ``` **After:** ```csharp public override void ConfigureIndex(CreateIndexRequestDescriptor idx) { base.ConfigureIndex(idx .Settings(s => s.NumberOfReplicas(0)) .Mappings(m => m.Properties(p => p.SetupDefaults()))); } ``` > **Note:** `AutoMap()` has been removed from the new client. Define all property mappings explicitly via `.Properties(...)`. ### ConfigureIndexMapping: Return Type and API `ConfigureIndexMapping` changed from returning `TypeMappingDescriptor` to `void`: | Before | After | |--------|-------| | `TypeMappingDescriptor ConfigureIndexMapping(TypeMappingDescriptor map)` | `void ConfigureIndexMapping(TypeMappingDescriptor map)` | | Returns the descriptor | Mutates the descriptor in place | **Before:** ```csharp public override TypeMappingDescriptor ConfigureIndexMapping(TypeMappingDescriptor map) { return map .Dynamic(false) .Properties(p => p .SetupDefaults() .Keyword(f => f.Name(e => e.Id)) .Text(f => f.Name(e => e.Name).AddKeywordAndSortFields()) ); } ``` **After:** ```csharp public override void ConfigureIndexMapping(TypeMappingDescriptor map) { map .Dynamic(DynamicMapping.False) .Properties(p => p .SetupDefaults() .Keyword(e => e.Id) .Text(e => e.Name, t => t.AddKeywordAndSortFields()) ); } ``` ### ConfigureIndexAliases Signature **Before:** ```csharp public override IPromise ConfigureIndexAliases(AliasesDescriptor aliases) { return aliases.Alias("my-alias"); } ``` **After:** ```csharp public override void ConfigureIndexAliases(FluentDictionaryOfNameAlias aliases) { aliases.Add("my-alias", a => a); } ``` ### CreateIndexAsync and UpdateIndexAsync Internal methods that create or update indexes changed from `Func` (fluent return) to `Action` (void mutation): | Before | After | |--------|-------| | `Func` | `Action` | | `Func` | `Action` | ### ConfigureSettings on Index **Before:** ```csharp public override void ConfigureSettings(ConnectionSettings settings) { } ``` **After:** ```csharp public override void ConfigureSettings(ElasticsearchClientSettings settings) { } ``` ## Property Mapping (TypeMappingDescriptor) Changes The new client uses a simpler expression syntax for property mappings. The `.Name(e => e.Prop)` wrapper is gone — property name inference comes directly from the expression. Configuration lambdas are now a second parameter: | Before | After | |--------|-------| | `.Keyword(f => f.Name(e => e.Id))` | `.Keyword(e => e.Id)` | | `.Text(f => f.Name(e => e.Name))` | `.Text(e => e.Name)` | | `.Text(f => f.Name(e => e.Name).Analyzer("my_analyzer"))` | `.Text(e => e.Name, t => t.Analyzer("my_analyzer"))` | | `.Number(f => f.Name(e => e.Age).Type(NumberType.Integer))` | `.IntegerNumber(e => e.Age)` | | `.Number(f => f.Name(e => e.Score).Type(NumberType.Double))` | `.DoubleNumber(e => e.Score)` | | `.Date(f => f.Name(e => e.CreatedUtc))` | `.Date(e => e.CreatedUtc)` | | `.Boolean(f => f.Name(e => e.IsActive))` | `.Boolean(e => e.IsActive)` | | `.Object(f => f.Name(e => e.Address).Properties(...))` | `.Object(e => e.Address, o => o.Properties(...))` | | `.Nested(f => f.Name(e => e.Items).Properties(...))` | `.Nested(e => e.Items, n => n.Properties(...))` | | `.Dynamic(false)` | `.Dynamic(DynamicMapping.False)` | | `.Map(m => m.Properties(...))` | `.Mappings(m => m.Properties(...))` | ### Number Type Mapping The generic `.Number()` with `NumberType` enum is replaced by specific typed methods: | Before | After | |--------|-------| | `.Number(f => f.Type(NumberType.Integer))` | `.IntegerNumber(e => e.Field)` | | `.Number(f => f.Type(NumberType.Long))` | `.LongNumber(e => e.Field)` | | `.Number(f => f.Type(NumberType.Float))` | `.FloatNumber(e => e.Field)` | | `.Number(f => f.Type(NumberType.Double))` | `.DoubleNumber(e => e.Field)` | ## Response Validation The `IsValid` property on responses was renamed to `IsValidResponse`: | Before | After | |--------|-------| | `response.IsValid` | `response.IsValidResponse` | | `response.OriginalException` | `response.OriginalException()` (method call) | | `response.ServerError?.Status` | `response.ElasticsearchServerError?.Status` | | `response.ServerError.Error.Type` | `response.ElasticsearchServerError.Error.Type` | ## Custom Field Type Mapping (ICustomFieldType) `ICustomFieldType.ConfigureMapping` changed from accepting a `SingleMappingSelector` parameter and returning `IProperty` to a parameterless method returning a factory function: **Before:** ```csharp public IProperty ConfigureMapping(SingleMappingSelector map) where T : class { return map.Number(n => n.Type(NumberType.Integer)); } ``` **After:** ```csharp public Func, IProperty> ConfigureMapping() where T : class { return factory => factory.IntegerNumber(); } ``` All standard field types (`IntegerFieldType`, `StringFieldType`, `BooleanFieldType`, `DateFieldType`, `KeywordFieldType`, `LongFieldType`, `FloatFieldType`, `DoubleFieldType`) have been updated to this pattern. If you have custom `ICustomFieldType` implementations, update them to match. ## Ingest Pipeline on Update The old client supported `Pipeline` on bulk update operations via a custom extension. **This feature is not supported by the Elasticsearch Update API** and has been removed. Use the Ingest pipeline on index (PUT) operations only. ## Snapshot API The `Snapshot.SnapshotAsync` method was renamed to `Snapshot.CreateAsync` in the new client. ## Counting with Index Filtering **Before:** ```csharp await client.CountAsync(d => d.Index(indexName), cancellationToken); ``` **After:** ```csharp await client.CountAsync(d => d.Indices(indexName)); ``` ## Parent-Child Documents The `RoutingField` configuration on `TypeMappingDescriptor` is no longer available as a direct mapping property. Routing is now handled at the index settings level or through query routing parameters. ## RefreshInterval **Before:** ```csharp settings.RefreshInterval(TimeSpan.FromSeconds(30)); ``` **After:** ```csharp settings.RefreshInterval(Duration.FromSeconds(30)); ``` ## TopHits Aggregation Round-Trip The `TopHitsAggregate` now serializes the raw document JSON in its `Hits` property, enabling round-trip serialization for caching purposes. The `Documents()` method checks both the in-memory `ILazyDocument` list (from a live ES response) and the serialized `Hits` list (from cache deserialization). ## Known Bugs and Workarounds ### ResolveIndexAsync in Elastic.Clients.Elasticsearch 9.x The `Indices.ResolveIndexAsync` method in Elastic.Clients.Elasticsearch 9.x is broken — it does not correctly resolve wildcard index patterns. Foundatio.Repositories works around this by using `Indices.GetAsync` with `IgnoreUnavailable()` instead: ```csharp // DON'T use ResolveIndexAsync — broken in Elastic.Clients.Elasticsearch 9.x // var resolved = await client.Indices.ResolveIndexAsync(pattern); // DO use GetAsync to resolve wildcard patterns var getResponse = await client.Indices.GetAsync( Indices.Parse("my-index-*"), d => d.IgnoreUnavailable()); if (getResponse.IsValidResponse && getResponse.Indices is not null) { foreach (var kvp in getResponse.Indices) Console.WriteLine(kvp.Key); // actual index name } ``` If you have code that calls `ResolveIndexAsync` directly, switch to `GetAsync`. ### EnableApiVersioningHeader Removed The `settings.EnableApiVersioningHeader()` call from NEST is no longer needed and does not exist in the new client. Remove it. ## Common Gotchas 1. **Fluent return vs void**: The most pervasive change is that descriptor-based methods (`ConfigureIndex`, `ConfigureIndexMapping`, `ConfigureIndexAliases`) no longer return the descriptor. Remove all `return` statements and change return types to `void`. 2. **AutoMap is gone**: The new client does not support `AutoMap()`. You must define every property mapping explicitly. This is actually safer — it prevents accidental mapping of fields you don't want indexed. 3. **Serializer mismatch**: If documents were serialized with Newtonsoft.Json (e.g., stored in a cache) and you try to deserialize with System.Text.Json, you may get errors or silent data loss. Ensure cached data is invalidated or re-serialized during migration. 4. **Enum serialization**: Both Newtonsoft.Json and System.Text.Json serialize enums as **integers** by default. `ConfigureFoundatioRepositoryDefaults()` does not register a global string-enum converter, and you usually need no extra attributes—existing indices that store enum values as integers stay compatible. 5. **Double precision**: System.Text.Json may round whole-number doubles (e.g., `1.0` becomes `1`). The `DoubleSystemTextJsonConverter` registered by `ConfigureFoundatioRepositoryDefaults()` preserves the decimal point, but only for `double` typed properties. 6. **object-typed properties**: Without `ObjectToInferredTypesConverter`, System.Text.Json deserializes `object` properties as `JsonElement` instead of CLR primitives. This converter is registered by `ConfigureFoundatioRepositoryDefaults()` but if you're using your own `JsonSerializerOptions`, you must add it manually. 7. **Indices.Parse vs IndexName cast**: When passing index names to API calls, use `(IndexName)name` for single names or `Indices.Parse(name)` for comma-separated or wildcard patterns. 8. **CancellationToken parameter changes**: Some API methods that previously accepted `CancellationToken` as a direct parameter now use it differently. Check each call site. 9. **OriginalException is a method**: `response.OriginalException` changed from a property to a method call `response.OriginalException()`. This will be a compile error, but it's easy to miss in string interpolation. 10. **ElasticsearchClientSettings is IDisposable**: The settings object implements `IDisposable` but hides it behind an explicit interface implementation. If you manage the client lifecycle yourself, cast to `IDisposable` and dispose it. ## Migration Checklist ### Packages and Namespaces * \[ ] Replace `using Elasticsearch.Net;` and `using Nest;` with `using Elastic.Clients.Elasticsearch;` * \[ ] Add additional namespaces as needed (`Mapping`, `IndexManagement`, `Aggregations`, etc.) * \[ ] Remove `NEST.JsonNetSerializer` dependency ### Configuration * \[ ] Update `CreateConnectionPool()` return type from `IConnectionPool` to `NodePool` * \[ ] Update pool class names (`SingleNodeConnectionPool` → `SingleNodePool`, etc.) * \[ ] Update `ConfigureSettings` parameter from `ConnectionSettings` to `ElasticsearchClientSettings` * \[ ] Update authentication calls (`.BasicAuthentication` → `.Authentication(new BasicAuthentication(...))`) * \[ ] Remove `settings.EnableApiVersioningHeader()` calls * \[ ] Pass an `ITextSerializer` to `ElasticConfiguration` if you need custom serialization ### Index Configuration * \[ ] Change `ConfigureIndex` return type from `CreateIndexDescriptor` to `void` (remove `return`) * \[ ] Change `ConfigureIndex` parameter from `CreateIndexDescriptor` to `CreateIndexRequestDescriptor` * \[ ] Change `ConfigureIndexMapping` return type to `void` (remove `return`) * \[ ] Change `ConfigureIndexAliases` to use `FluentDictionaryOfNameAlias` and `void` return * \[ ] Replace `.Map(...)` with `.Mappings(...)` * \[ ] Remove `AutoMap()` calls; define all mappings explicitly ### Property Mappings * \[ ] Update property mapping syntax (remove `.Name(e => e.Prop)` wrapper) * \[ ] Replace `.Number(f => f.Type(NumberType.Integer))` with `.IntegerNumber(e => e.Field)` * \[ ] Replace `.Dynamic(false)` with `.Dynamic(DynamicMapping.False)` * \[ ] Update `.Text()`, `.Object()`, `.Nested()` to use two-parameter form for configuration ### Serialization * \[ ] Replace `[JsonProperty]` (Newtonsoft) with `[JsonPropertyName]` (System.Text.Json) * \[ ] Rewrite custom `JsonConverter` classes for System.Text.Json * \[ ] Use `ConfigureFoundatioRepositoryDefaults()` on your `JsonSerializerOptions` * \[ ] Update `LazyDocument` construction to provide a required `ITextSerializer` * \[ ] Invalidate caches that may contain Newtonsoft-serialized data ### Response Handling * \[ ] Replace `response.IsValid` with `response.IsValidResponse` * \[ ] Replace `response.OriginalException` with `response.OriginalException()` (method call) * \[ ] Replace `response.ServerError` with `response.ElasticsearchServerError` ### Custom Field Types * \[ ] Update `ICustomFieldType.ConfigureMapping` to new `Func, IProperty>` signature ### Known Issues * \[ ] Replace any `ResolveIndexAsync` calls with `Indices.GetAsync` (broken in Elastic.Clients.Elasticsearch 9.x) * \[ ] Verify enum serialization compatibility with existing Elasticsearch data * \[ ] Test document round-tripping with System.Text.Json --- --- url: /guide/consistency.md --- # Consistency and Dirty Reads Elasticsearch uses a **near real-time** search model. Understanding when your reads are real-time vs. eventually consistent is critical to avoiding subtle concurrency bugs. ## How Elasticsearch Segments Work When you index (write) a document, Elasticsearch immediately writes it to the **transaction log** (translog) and an in-memory buffer. The document is retrievable by ID at this point through the realtime GET path. However, it is not *searchable* until the next **refresh**, which flushes the buffer into a searchable **segment** (default: every 1 second). ```mermaid sequenceDiagram participant App participant ES as Elasticsearch participant TLog as Transaction Log participant Seg as Searchable Segments App->>ES: Index document ES->>TLog: Write immediately Note over TLog: Available for GET by ID App->>ES: GetByIdAsync ES->>TLog: Realtime read TLog-->>App: Latest document App->>ES: FindAsync / ExistsAsync(query) ES->>Seg: Search (near real-time) Seg-->>App: May not include recent writes Note over ES,Seg: Refresh (~1s) ES->>Seg: Flush buffer to segment Note over Seg: Now searchable ``` This creates two fundamentally different read paths: * **GET path** (real-time): Reads directly from the transaction log. Returns the latest version of a document immediately after a write, even before a refresh. * **Search path** (near real-time): Queries the searchable segments, which lag behind writes by up to the refresh interval. Writes that haven't been refreshed yet are invisible -- this is a **dirty read**. ## Repository Operations by Consistency | Operation | Real-Time? | ES API Used | Notes | |-----------|------------|-------------|-------| | `GetByIdAsync` | ✅ Yes | GET API | Falls back to Search when model has a parent and no routing is provided | | `GetByIdsAsync` | ✅ Yes | Multi-GET API | Falls back to Search for unrouted parent documents or multi-index | | `ExistsAsync(id)` | ✅ Yes | GET API / Document Exists API | Uses Document Exists API (no soft deletes) or GET API with source filter (soft deletes). Falls back to Search only for unrouted parent documents | | `ExistsAsync(query)` | ❌ No | Search API (`size: 0`) | Always Search, even with `.Id(id)` combined with field filters | | `FindAsync` | ❌ No | Search API | Subject to refresh interval | | `FindOneAsync` | ❌ No | Search API (`size: 1`) | Subject to refresh interval | | `CountAsync` | ❌ No | Search API (`size: 0`) | Uses Search (not the Count API) to support aggregations | | `GetAllAsync` | ❌ No | Search API | Delegates to `FindAsync` with an empty query | | `BatchProcessAsync` | ❌ No | Search API | Iterates with search-after paging via `FindAsAsync` | ## The Dirty Read Problem Any method using the search path can return stale results during the refresh window: ```csharp var employee = await repository.AddAsync(new Employee { CompanyId = "company-123", Name = "Jane Doe" }); // Search path -- might NOT find the employee yet (dirty read) var hit = await repository.FindOneAsync(q => q.FieldEquals(e => e.CompanyId, employee.CompanyId)); // hit.Document could be null! // GET path -- WILL find it immediately (real-time) var byId = await repository.GetByIdAsync(employee.Id); // byId is guaranteed to be the latest version ``` ## Common Pitfalls ### ExistsAsync with Field Filters `ExistsAsync(id)` uses the real-time Document Exists API, but `ExistsAsync(query)` always uses the search path -- even when the query targets a specific ID. Adding any field filter forces the query overload: ```csharp employee.EmploymentType = EmploymentType.Contract; await repository.SaveAsync(employee); // Search path -- the index hasn't refreshed yet bool isContract = await repository.ExistsAsync(q => q .Id(employee.Id) .FieldEquals(e => e.EmploymentType, EmploymentType.Contract)); // isContract could be false! // GET path -- real-time, always accurate var fresh = await repository.GetByIdAsync(employee.Id, o => o.Include(e => e.EmploymentType)); bool freshIsContract = fresh is not null && fresh.EmploymentType == EmploymentType.Contract; ``` ### ExistsAsync(id) with Soft Deletes When a model implements `ISupportSoftDeletes`, `ExistsAsync(id)` uses the real-time GET API with a source filter to fetch only the `IsDeleted` field, then checks it in code. This provides real-time accuracy with minimal payload: ```csharp // Employee implements ISupportSoftDeletes employee.IsDeleted = true; await repository.SaveAsync(employee); // Real-time: uses GET API with _source_includes=isDeleted, then checks IsDeleted in code bool exists = await repository.ExistsAsync(employee.Id); // exists is false -- the soft deletion is visible immediately via the GET path ``` ## Solving Dirty Reads ### When You Have the Document ID If you know the document ID and need to check a field's current state, use `GetByIdAsync` -- it reads from the transaction log and is always consistent. Use `Include` to fetch only the fields you need: ```csharp var employee = await repository.GetByIdAsync(id, o => o.Include(e => e.EmploymentType)); bool isContract = employee is not null && employee.EmploymentType == EmploymentType.Contract; ``` This replaces patterns like `ExistsAsync(q => q.Id(id).FieldEquals(...))` and avoids the search path entirely. For simple existence checks without field filters, `ExistsAsync(id)` is already real-time. ### When You're Searching by Field When you need to look up documents by a non-ID field (e.g., email, company, slug), the search path is unavoidable. Use custom cache keys to make these lookups reliable across the refresh window: ```csharp public class UserRepository : ElasticRepositoryBase { public async Task GetByEmailAddressAsync(string emailAddress) { if (String.IsNullOrWhiteSpace(emailAddress)) return null; emailAddress = emailAddress.Trim().ToLowerInvariant(); var hit = await FindOneAsync( q => q.FieldEquals(u => u.EmailAddress, emailAddress), o => o.Cache($"email:{emailAddress}")); return hit?.Document; } } ``` The first call searches Elasticsearch (may be a dirty read), but caches the result by the email key. Subsequent calls return the cached result. When the document is saved, the repository's cache invalidation clears the key, and the next lookup fetches fresh data. See [Caching - Custom Cache Keys for Eventual Consistency](caching.md#custom-cache-keys-for-eventual-consistency) for the full pattern with `InvalidateCacheAsync`. ### ImmediateConsistency (Tests Only) `ImmediateConsistency()` forces an Elasticsearch index refresh, making the search path consistent. **Never use this in production** -- it degrades cluster performance. You can apply it to either the write or the read: ```csharp // Force refresh on the write -- all subsequent searches see the update await repository.SaveAsync(employee, o => o.ImmediateConsistency()); // Or force refresh on the read -- only this search is guaranteed consistent bool exists = await repository.ExistsAsync(q => q .Id(employee.Id) .FieldEquals(e => e.EmploymentType, EmploymentType.Contract), o => o.ImmediateConsistency()); ``` ## Consistency Modes (Internal Mechanics) The `Consistency` enum controls how the repository ensures write visibility: | Mode | Write Behavior | Read Behavior | Use Case | |------|---------------|---------------|----------| | `Eventual` (default) | `Refresh.False` -- no forced refresh | No pre-query refresh | Normal production operations | | `Immediate` | `Refresh.True` -- synchronous refresh before response | `Indices.RefreshAsync` before search | Tests, critical paths needing instant visibility | | `Wait` | `Refresh.WaitFor` -- blocks until next scheduled refresh | `Indices.RefreshAsync` before search | Lower-overhead alternative to Immediate | ### Setting Consistency ```csharp // On a specific operation await repository.AddAsync(doc, o => o.ImmediateConsistency()); // Or wait for next scheduled refresh (gentler on the cluster) await repository.AddAsync(doc, o => o.ImmediateConsistency(shouldWait: true)); ``` ### DefaultConsistency Property Repositories can set `DefaultConsistency` in the constructor to apply a non-Eventual mode by default for all operations: ```csharp public class MigrationStateRepository : ElasticRepositoryBase { public MigrationStateRepository(IIndex index) : base(index) { DefaultConsistency = Consistency.Immediate; } } ``` This is appropriate for small, correctness-critical indices (migration state, field definitions). Avoid for large or high-write indices due to the refresh cost. ### How Read-Side Refresh Works Search-based operations (`FindAsync`, `CountAsync`, `ExistsAsync(query)`, `FindOneAsync`) call `RefreshForConsistency` before executing the search. This is essential for: 1. **Cross-process consistency**: Process A writes with Eventual consistency, Process B reads with Immediate -- the read-side refresh ensures B sees A's write. 2. **Batch pagination correctness**: `RemoveAllAsync` and `PatchAllAsync` intentionally downgrade inner writes to Eventual for performance, then rely on the read-side refresh between pages to prevent reprocessing. ### Batch Processing and Consistency `BatchProcessAsAsync` (used by `RemoveAllAsync` and `PatchAllAsync` when listeners/cache are active) uses this pattern: 1. Refresh (via `FindAsAsync`) → search for page 1 2. Process batch (inner writes use Eventual for performance) 3. Refresh (via next `FindAsAsync`) → search for page 2 (correctly skips processed docs) 4. Repeat until done 5. Final `RefreshForConsistency` for the last batch The inner writes are downgraded to Eventual to avoid the cost of per-write refresh during bulk operations. The inter-page refresh makes the completed batch invisible to subsequent queries, preventing infinite loops or double-processing. ### Known Limitations * **`UpdateByQuery` and `DeleteByQuery`**: These ES APIs only accept `bool?` for their refresh parameter (not the `Refresh` enum). `Consistency.Wait` is treated identically to `Immediate` for these operations. * **Non-idempotent scripts**: If using `PatchAllAsync` with a non-idempotent script (e.g., counter increment) and Eventual consistency, a document could theoretically be processed twice between page refreshes. Use `ImmediateConsistency` for non-idempotent patches. * **Exceptions bypass final refresh**: If an exception occurs during batch processing, the final refresh is skipped. Partially-processed writes may not be immediately visible. ## Next Steps * [Caching](caching.md) - How the cache layer handles dirty reads * [Querying](querying.md) - Query syntax for search-based operations * [CRUD Operations](crud-operations.md) - Complete operations reference * [Troubleshooting](troubleshooting.md) - Common issues and solutions --- --- url: /README.md --- # Foundatio.Repositories Documentation This folder contains the VitePress documentation site for Foundatio.Repositories. ## Development ```bash # Install dependencies npm install # Start development server npm run dev # Build for production npm run build # Preview production build npm run preview ``` ## Structure * `.vitepress/config.ts` - VitePress configuration * `guide/` - Documentation pages * `public/` - Static assets * `index.md` - Homepage ## Deployment Documentation is automatically deployed to GitHub Pages when changes are pushed to the `main` branch. See `.github/workflows/docs.yml` for the deployment workflow.