# Comparison import { Callout } from 'fumadocs-ui/components/callout'; This page compares TurboMediator against the two most referenced .NET libraries in this space: **MediatR** (the original mediator library) and **MassTransit** (the distributed messaging framework). These libraries solve different problems. MassTransit is a distributed message bus — comparing it here is useful because teams sometimes reach for it when they only need in-process messaging. MediatR is the closest direct competitor. *** Feature Matrix [#feature-matrix] | Feature | MediatR | MassTransit | TurboMediator | | ----------------------------------------- | :-------: | :---------: | :-----------: | | **Core** | | | | | Request / Response | ✅ | ✅ | ✅ | | Notifications / Events | ✅ | ✅ | ✅ | | Streaming | ✅ | ✅ | ✅ | | Pipeline behaviors | ✅ | ✅ | ✅ | | **Performance** | | | | | Source-generated dispatch (no reflection) | ❌ | ❌ | ✅ | | Native AOT support | ❌ | Partial | ✅ | | Compile-time missing handler detection | ❌ | ❌ | ✅ | | Zero-overhead in-process dispatch | ❌ | ❌ | ✅ | | **Cross-cutting Concerns** | | | | | Built-in caching | ❌ | ❌ | ✅ | | Built-in validation | ❌ | ❌ | ✅ | | FluentValidation integration | Community | ❌ | ✅ | | Rate limiting | ❌ | ❌ | ✅ | | Distributed locking | ❌ | ❌ | ✅ | | Feature flags | ❌ | ❌ | ✅ | | **Resilience** | | | | | Retry | ❌ | ✅ | ✅ | | Circuit breaker | ❌ | ✅ | ✅ | | Timeout | ❌ | ✅ | ✅ | | Fallback | ❌ | ❌ | ✅ | | Hedging | ❌ | ❌ | ✅ | | **Observability** | | | | | OpenTelemetry tracing | Community | ✅ | ✅ | | Metrics | Community | ✅ | ✅ | | Structured logging | Community | ✅ | ✅ | | Correlation ID propagation | ❌ | ✅ | ✅ | | Health checks | ❌ | ✅ | ✅ | | **Workflows** | | | | | Sagas | ❌ | ✅ | ✅ | | Saga persistence (EF Core) | ❌ | ✅ | ✅ | | State machines | ❌ | ✅ | ✅ | | Scheduling / Recurring jobs | ❌ | ✅ | ✅ | | **Persistence** | | | | | Transactional outbox | ❌ | ✅ | ✅ | | Inbox (deduplication) | ❌ | ✅ | ✅ | | Audit trail | ❌ | ❌ | ✅ | | EF Core integration | ❌ | ✅ | ✅ | | **Enterprise** | | | | | Authorization | ❌ | ❌ | ✅ | | Multi-tenancy | ❌ | ❌ | ✅ | | Message deduplication | ❌ | ✅ | ✅ | | **Distributed Messaging** | | | | | RabbitMQ, Azure Service Bus, etc. | ❌ | ✅ | ❌ | | Multiple transport adapters | ❌ | ✅ | ❌ | | **Developer Experience** | | | | | CLI tooling | ❌ | ❌ | ✅ | | Testing utilities | ✅ | ✅ | ✅ | | **Licensing** | | | | | Free for all teams | ❌¹ | ❌² | ✅ | ¹ MediatR's Community license is free only for companies/individuals with less than $5M USD in annual gross revenue that have never received more than $10M in outside capital. Source: [mediatr.io](https://mediatr.io). ² MassTransit v9 transitioned to a commercial licensing model. v8 remains open-source but will only receive security patches until end of 2026. Pricing starts at $400/month for small/medium businesses. Source: [MassTransit v9 announcement](https://masstransit.io/introduction/v9-announcement). *** When to Choose What [#when-to-choose-what] Choose TurboMediator when [#choose-turbomediator-when] * You need **in-process messaging** with the fastest possible dispatch (source-generated, no reflection) * You want **AOT / trimming** support * You want a **complete pipeline ecosystem** — resilience, validation, caching, sagas, scheduling, observability — all built to work together * You care about **compile-time safety** (missing handlers are caught at build time) * You need **enterprise features** like authorization, multi-tenancy, or an audit trail without stitching together separate libraries Choose MediatR when [#choose-mediatr-when] * You have a **simple project** that only needs request/response dispatch and pipeline behaviors * Your team is already familiar with it and the use case doesn't justify a migration * Budget and licensing are not a concern Choose MassTransit when [#choose-masstransit-when] * You need **distributed messaging** across services (RabbitMQ, Azure Service Bus, Amazon SQS, etc.) * Your architecture is **event-driven across process boundaries** rather than within a single process * You need **distributed transaction choreography** with Routing Slips MassTransit and TurboMediator are **not mutually exclusive**. A common architecture uses MassTransit for cross-service messaging and TurboMediator for in-process command/query dispatch within each service. *** Performance Difference [#performance-difference] The main performance differentiator is how message dispatch works under the hood. | Approach | Dispatch mechanism | Runtime cost | | ------------------------ | ---------------------------------------------- | -------------------------------- | | MediatR | Reflection + `Dictionary` lookup | Allocation + reflection per call | | MassTransit (in-process) | Expression-compiled delegates | One-time compilation cost | | TurboMediator | Source-generated `switch` expression | Zero — resolved at compile time | TurboMediator's source generator emits code like this at build time: ```csharp // Generated — no reflection, no dictionary lookup public ValueTask Send(IRequest request, CancellationToken ct) => request switch { CreateOrderCommand r => (ValueTask)(object)_createOrderHandler.Handle(r, ct), GetOrderQuery r => (ValueTask)(object)_getOrderHandler.Handle(r, ct), // ... all registered handlers _ => ThrowUnknownRequest(request) }; ``` This means **no heap allocations** for the dispatch itself and full **inlining** by the JIT. # TurboMediator import { Cards, Card } from 'fumadocs-ui/components/card'; import { Callout } from 'fumadocs-ui/components/callout'; TurboMediator is currently in **preview**. APIs may change between releases without prior notice. It is usable in production, but exercise caution: thoroughly test in your environment, pin to an exact version in your dependencies, and review the changelog before upgrading. What is TurboMediator? [#what-is-turbomediator] TurboMediator is a **high-performance implementation** of the Mediator pattern for .NET that uses **Roslyn Source Generators** to create compile-time optimized dispatch code. Unlike traditional mediator libraries that rely on reflection and runtime type resolution, TurboMediator generates all the routing code at build time. Key Benefits [#key-benefits] * **Zero Reflection** — All handler dispatch is generated at compile time via `switch` expressions * **Native AOT Compatible** — No runtime code generation, fully compatible with Native AOT publishing * **Compile-Time Validation** — Missing handlers, duplicate handlers, and signature mismatches are caught as build errors * **ValueTask-Based** — Uses `ValueTask` throughout for minimal allocation overhead * **CQRS Ready** — Built-in support for Commands, Queries, Requests, and Notifications * **Streaming** — First-class `IAsyncEnumerable` support for streaming scenarios * **Modular** — 20+ optional packages for cross-cutting concerns Package Overview [#package-overview] | Package | Description | | ---------------------------------------------- | ------------------------------------------------------------------------ | | `TurboMediator.Abstractions` | Core interfaces, message types, handlers, pipeline behaviors | | `TurboMediator.SourceGenerator` | Compile-time code generation (required) | | `TurboMediator.FluentValidation` | FluentValidation integration | | `TurboMediator.Resilience` | Retry, Circuit Breaker, Timeout, Fallback, Hedging | | `TurboMediator.Result` | Result pattern types for functional error handling | | `TurboMediator.Observability` | Telemetry, Metrics, Correlation, Structured Logging, Health Checks | | `TurboMediator.Caching` | Response caching with attribute-based configuration | | `TurboMediator.Caching.Redis` | Redis distributed cache provider for TurboMediator.Caching | | `TurboMediator.Validation` | Built-in lightweight validation (no FluentValidation dependency) | | `TurboMediator.Enterprise` | Authorization, Multi-Tenancy, Deduplication | | `TurboMediator.Scheduling` | Cron jobs and recurring job scheduling | | `TurboMediator.Scheduling.EntityFramework` | EF Core job store for Scheduling | | `TurboMediator.RateLimiting` | Rate Limiting & Bulkhead Isolation | | `TurboMediator.Persistence` | Transactions, Outbox, Audit | | `TurboMediator.Persistence.EntityFramework` | Entity Framework Core implementations | | `TurboMediator.Saga` | Saga orchestration pattern | | `TurboMediator.Saga.EntityFramework` | EF Core saga store | | `TurboMediator.StateMachine` | State machine pattern with guards, transitions, and mediator integration | | `TurboMediator.StateMachine.EntityFramework` | EF Core transition store for state machine auditing | | `TurboMediator.Batching` | Batch processing for queries | | `TurboMediator.FeatureFlags` | Feature flag behavior | | `TurboMediator.FeatureFlags.FeatureManagement` | Microsoft.FeatureManagement integration | | `TurboMediator.Testing` | Test utilities, fakes, and helpers | | `TurboMediator.Cli` | CLI tool for analysis and docs generation | Next Steps [#next-steps] # Vibecoded with Opus import { Callout } from 'fumadocs-ui/components/callout'; Yes, This Library Is Vibecoded [#yes-this-library-is-vibecoded] TurboMediator was built entirely through **AI-assisted development** — specifically using [Claude Opus](https://www.anthropic.com/claude), Anthropic's most capable model. Every module, every abstraction, every source generator, every pipeline behavior was authored through a collaborative loop between human intent and AI generation. There was no traditional keyboard-driven coding session. The architecture was described, the constraints were stated, the tradeoffs were discussed — and the code emerged from that conversation. This is **vibecoding**: directing software into existence through high-bandwidth dialogue with a model capable of holding an entire codebase's worth of context. *** Why That Should Concern You (and Why It Doesn't) [#why-that-should-concern-you-and-why-it-doesnt] The reasonable reaction to hearing a library is vibecoded is skepticism. AI models hallucinate. They produce plausible-looking code that subtly violates invariants. They miss edge cases that a battle-scarred engineer would catch in seconds. They optimize for code that *reads* correct, not code that *is* correct. These are legitimate concerns. But they apply to **the generation process**, not the **artifact**. The test suite is the equalizer. *** The Test Suite [#the-test-suite] TurboMediator has a comprehensive suite of **unit and integration tests** covering every layer of the library: Unit Tests [#unit-tests] Every core behavior is tested in isolation: | Area | Test File | | ------------------------- | ------------------------------------------------------- | | Commands | `CommandTests.cs` | | Queries | `QueryTests.cs` | | Requests | `RequestTests.cs` | | Notifications | `NotificationTests.cs`, `NotificationPublisherTests.cs` | | Pipeline behaviors | `PipelineBehaviorTests.cs` | | Exception handling | `ExceptionHandlerTests.cs` | | Streaming | `StreamingTests.cs` | | Dependency injection | `DependencyInjectionTests.cs` | | Validation | `ValidationTests.cs`, `FluentValidationTests.cs` | | Caching | `CachingTests.cs` | | Batching | `BatchingTests.cs` | | Rate limiting | `RateLimiting/` | | Resilience | `ResilienceTests.cs`, `Resilience/` | | Feature flags | `FeatureFlagTests.cs` | | Result types | `ResultTests.cs` | | Sagas | `SagaTests.cs` | | State machines | `StateMachineTests.cs` | | Telemetry / Observability | `TelemetryTests.cs`, `Observability/` | | Distributed locking | `DistributedLocking/` | | Inbox / Outbox | `Inbox/`, `Outbox/` | | Processors | `ProcessorTests.cs` | | Enterprise features | `Enterprise/` | | Scheduling | `Scheduling/` | | CLI | `Cli/` | | Testing utilities | `Testing/` | Integration Tests [#integration-tests] End-to-end scenarios test real interactions between components, including database-backed stores and external infrastructure: | Test | What it covers | | ----------------------------------------- | --------------------------------------------------------------- | | `FullPipelineIntegrationTests.cs` | A complete request flowing through all pipeline behaviors | | `CachingIntegrationTests.cs` | Distributed cache read/write/invalidation under a real pipeline | | `TransactionIntegrationTests.cs` | Unit-of-work and transactional consistency with EF Core | | `AuditStoreIntegrationTests.cs` | Audit trail persistence for commands | | `SagaStoreIntegrationTests.cs` | Saga state persistence and resumption | | `InboxStoreIntegrationTests.cs` | Inbox idempotency guarantees | | `OutboxStoreIntegrationTests.cs` | Outbox reliable delivery semantics | | `RedisDistributedLockIntegrationTests.cs` | Distributed locking via Redis | AOT Compatibility [#aot-compatibility] A dedicated `AotCompatibilityTests.cs` verifies that the source-generated output is fully compatible with **Native AOT** — no reflection, no runtime code emission, no trimming surprises. *** Benchmark Results [#benchmark-results] Numbers don't care who wrote the code. These results were produced by BenchmarkDotNet running against TurboMediator, MediatR, and the [Mediator](https://github.com/martinothamar/Mediator) source-generator library (the closest architectural equivalent). > Intel Core i7-10750H, .NET 8.0.23, X64 RyuJIT AVX2 Send Command [#send-command] | Method | Mean | Ratio | Allocated | Alloc Ratio | | ------------------- | --------: | -----------: | --------: | ----------: | | TurboMediator | 55.56 ns | baseline | 24 B | | | Mediator\_SourceGen | 55.91 ns | 1.01x slower | 24 B | 1.00x more | | MediatR | 219.16 ns | 3.94x slower | 296 B | 12.33x more | Send Query [#send-query] | Method | Mean | Ratio | Allocated | Alloc Ratio | | ------------------- | --------: | -----------: | --------: | ----------: | | TurboMediator | 77.29 ns | baseline | 48 B | | | Mediator\_SourceGen | 68.36 ns | 1.13x faster | 48 B | 1.00x more | | MediatR | 235.85 ns | 3.05x slower | 464 B | 9.67x more | Send Command With Response [#send-command-with-response] | Method | Mean | Ratio | Allocated | Alloc Ratio | | ------------------- | --------: | -----------: | --------: | ----------: | | TurboMediator | 72.76 ns | baseline | 48 B | | | Mediator\_SourceGen | 66.50 ns | 1.09x faster | 48 B | 1.00x more | | MediatR | 248.88 ns | 3.42x slower | 464 B | 9.67x more | Pipeline Behavior (validation + logging + caching behaviors active) [#pipeline-behavior-validation--logging--caching-behaviors-active] | Method | Mean | Ratio | Allocated | Alloc Ratio | | ------------------- | --------: | -----------: | --------: | ----------: | | TurboMediator | 74.81 ns | baseline | 48 B | | | Mediator\_SourceGen | 78.64 ns | 1.05x slower | 48 B | 1.00x more | | MediatR | 316.44 ns | 4.23x slower | 656 B | 13.67x more | Publish Notification [#publish-notification] | Method | Mean | Ratio | Allocated | Alloc Ratio | | ------------------- | --------: | -----------: | --------: | ----------: | | Mediator\_SourceGen | 57.20 ns | 1.48x faster | 24 B | 1.00x more | | TurboMediator | 84.40 ns | baseline | 24 B | | | MediatR | 216.04 ns | 2.56x slower | 344 B | 14.33x more | TurboMediator is **3–4× faster than MediatR** and allocates **10–14× less memory** per operation. It runs at parity with the other source-generator-based implementation. This is what compile-time dispatch with zero reflection produces — and the AI wrote every line of it. *** How the Development Process Worked [#how-the-development-process-worked] Vibecoding is not "paste requirements, accept all." The loop looked like this: 1. **Describe the behavior** — state the interface contract, the invariants, the failure modes, and the performance constraints in precise terms 2. **Generate** — the model produces an implementation against those constraints 3. **Run the tests** — the test suite either passes or reveals where the model's mental model diverged from reality 4. **Review and challenge** — every non-trivial design decision was questioned: *why this abstraction? what does this break? is this the right tradeoff?* 5. **Iterate** — the loop repeats, refining until the tests pass and the design holds up under scrutiny No code entered the codebase without passing the test suite. No design decision was accepted without a rationale that survived a challenge. The model generates fast; the human stays critical. *** How It Compares [#how-it-compares] If the vibecoded origin still bothers you, the [comparison page](/docs/comparison) benchmarks TurboMediator against MediatR feature-by-feature — licensing, performance, Native AOT support, pipeline capabilities, and ecosystem coverage. The argument against vibecoded code is that it might be subtly wrong. The comparison shows what "subtly wrong" would have to overcome: a 4× performance advantage, zero allocations in the hot path, Native AOT compatibility, and 20+ modules — all with a passing test suite. *** The Point [#the-point] Code quality is not a function of who — or what — wrote it. It is a function of how thoroughly it is **specified**, **reviewed**, and **tested**. Every behavior in this library carries a test. Every edge case that was identified during development was encoded as an assertion. The test suite is the specification made executable. An AI that produces a correct, well-tested implementation is more useful than a human who produces an untested one. The output is what matters, and the output can be verified. Open an issue. That's exactly what open source exists for — regardless of who wrote the code. The repository is public, the tests are the ground truth, and a reproducible bug report is always welcome. This page exists because hiding the development process would be dishonest. Every module was reviewed critically before being accepted — the AI generated, the human decided. If you want to audit a specific design decision, the source is open. # Why TurboMediator? import { Callout } from 'fumadocs-ui/components/callout'; The Elephant in the Room [#the-elephant-in-the-room] In early 2025, within days of each other, both MediatR and MassTransit announced their transition to commercial licensing models. The two most widely-used .NET messaging libraries going paid almost simultaneously sent shockwaves through the community. A wave of blog posts and conference talks argued you should **stop using the Mediator pattern altogether**. The most common reactions are: * *"ASP.NET Core already has middleware and filters."* * *"It's just an unnecessary abstraction over method calls."* * *"You can call your services directly."* These arguments are not wrong — they're just **incomplete**. They focus on a narrow slice of .NET development (small Web API projects) and ignore the reality of large, complex systems where the pattern truly shines. Let's understand **why** this perception exists before we break down the specific arguments. *** Why People Don't See the Value [#why-people-dont-see-the-value] The dismissal of the Mediator pattern isn't random — it follows a predictable pattern rooted in how the community was exposed to it. 1. Every tutorial shows a TODO app [#1-every-tutorial-shows-a-todo-app] The vast majority of MediatR tutorials demonstrate the pattern with a trivial CRUD API: create a `CreateTodoCommand`, wire up a handler, and... that's it. In that context, the pattern genuinely **is** unnecessary overhead. You added an interface, a handler class, and a pipeline — to call `_db.Add(todo)`. The problem isn't the pattern. The problem is that **no one shows the pipeline**. The real power of the Mediator pattern — behaviors for validation, logging, caching, transactions, retry policies, telemetry — is almost never demonstrated in introductory content. People form their opinion based on the simplest possible example and never see the 90% of value that comes from the pipeline. 2. Paid licensing triggered a backlash against the pattern itself [#2-paid-licensing-triggered-a-backlash-against-the-pattern-itself] When MediatR and MassTransit announced commercial licenses within days of each other in April 2025, the community reaction wasn't just "let's find an alternative." A significant portion of developers took it as an opportunity to argue the patterns themselves were unnecessary. The reasoning went: *"If I have to pay for it, maybe I never needed it."* The simultaneity made the backlash louder than it would have been otherwise — two separate communities reacting at the same time, feeding the same narrative. But this conflates the **licensing model of specific libraries** with the **usefulness of architectural patterns**. The Mediator pattern and event-driven messaging existed long before these libraries and will exist long after. Licensing decisions are business decisions — they say nothing about whether a pattern solves a real problem. 3. People generalize from their own context [#3-people-generalize-from-their-own-context] A developer building a 5-endpoint API with no background workers, no event consumers, and no complex cross-cutting concerns will correctly conclude that the pattern adds no value **for them**. The mistake is assuming that experience applies universally. In large codebases with multiple entry points, dozens of developers, and strict requirements around auditability, resilience, and consistency — the pattern pays for itself many times over. But you can't see that from the outside looking in. 4. The "just call the service directly" trap [#4-the-just-call-the-service-directly-trap] Direct service injection feels simpler because the complexity is **hidden**, not absent. In a small project, `controller → service → repository` works fine. But as the application grows, each service method gradually accumulates validation, logging, authorization checks, caching logic, and error handling — scattered inconsistently across the codebase. The Mediator pattern makes this complexity **explicit and centralized**. It feels like more ceremony upfront because it forces you to name your operations and define your pipeline. That's not a weakness — it's the entire point. 5. Confusion between the pattern and the library [#5-confusion-between-the-pattern-and-the-library] Many critiques of "the Mediator pattern" are actually critiques of a specific library's API or implementation choices. Service location concerns, runtime reflection, dictionary lookups — these are **implementation details**, not properties of the pattern itself. A source-generated mediator (like TurboMediator) has none of these issues. The dispatch is a compile-time `switch` expression. There's no service locator. There's no reflection. Dismissing the pattern because of how one library implemented it is like dismissing dependency injection because one container was slow. *** Now let's address the specific technical arguments. *** "ASP.NET middleware is enough" [#aspnet-middleware-is-enough] ASP.NET middleware and endpoint filters are powerful — **for HTTP request pipelines**. But what about: * **Background workers** (`IHostedService`, `BackgroundService`) processing messages from queues? * **Scheduled jobs** (Hangfire, Quartz.NET, custom cron jobs)? * **gRPC services**? * **SignalR hubs**? * **Console applications** and CLI tools? * **Event-driven architectures** where messages flow between bounded contexts? In a real-world enterprise system, your business logic doesn't live exclusively behind HTTP endpoints. You often have **multiple entry points** — an API, several workers, scheduled tasks, event consumers — all executing the **same business operations**. ASP.NET middleware **only covers HTTP**. The Mediator pipeline covers **everything**. ``` ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Web API │ │ Worker │ │ Cron Job │ │ gRPC/Hub │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ └────────────────┴────────────────┴────────────────┘ │ ┌───────────▼───────────┐ │ Mediator Pipeline │ │ ┌─────────────────┐ │ │ │ Validation │ │ │ │ Logging │ │ │ │ Authorization │ │ │ │ Caching │ │ │ │ Transactions │ │ │ │ Retry/CB │ │ │ │ Telemetry │ │ │ └─────────────────┘ │ │ │ │ │ ┌──────▼──────┐ │ │ │ Handler │ │ │ └─────────────┘ │ └───────────────────────┘ ``` When you have a `CreateOrderCommand`, it doesn't matter if it was triggered by an API endpoint, a message from RabbitMQ, or a scheduled job. The **same pipeline behaviors** (validation, authorization, logging, transactions, telemetry) are applied consistently. Without a mediator, you'd need to **duplicate** those cross-cutting concerns across every entry point — or build your own pipeline abstraction, which is exactly what a mediator is. *** "It's just an extra layer of indirection" [#its-just-an-extra-layer-of-indirection] Yes. That's the point. Every useful pattern in software adds a layer of indirection in exchange for something valuable. Dependency injection is indirection. Interfaces are indirection. The question is: **what do you get in return?** With the Mediator pattern, you get: 1. Decoupled Architecture [#1-decoupled-architecture] Your controllers, workers, and jobs don't need to know which service handles what. They send a message and receive a response. This makes your code **easier to refactor**, **easier to test**, and **easier to reason about**. ```csharp // Controller doesn't know or care who handles this public async Task CreateOrder([FromBody] CreateOrderRequest dto) { var result = await mediator.Send(new CreateOrderCommand(dto.CustomerId, dto.Items)); return Ok(result); } // Worker doesn't know or care who handles this public async Task ProcessMessage(QueueMessage msg) { var command = JsonSerializer.Deserialize(msg.Body); await mediator.Send(command); } ``` Both entry points share the **same command**, the **same handler**, and the **same pipeline** — zero duplication. 2. Uniform Cross-Cutting Concerns [#2-uniform-cross-cutting-concerns] Without a mediator, here's what adding logging to every operation looks like: ```csharp // ❌ Without mediator — you repeat this in every service method public async Task CreateOrder(CreateOrderCommand command) { _logger.LogInformation("Creating order for {CustomerId}", command.CustomerId); var sw = Stopwatch.StartNew(); try { // validate... // authorize... // actual logic... _logger.LogInformation("Order created in {Elapsed}ms", sw.ElapsedMilliseconds); return order; } catch (Exception ex) { _logger.LogError(ex, "Failed to create order"); throw; } } ``` With a mediator pipeline, you write it **once**: ```csharp // ✅ With mediator — write once, apply to every operation public class LoggingBehavior : IPipelineBehavior where TRequest : IMessage { public async ValueTask Handle( TRequest request, MessageHandlerDelegate next, CancellationToken ct) { _logger.LogInformation("Handling {Request}", typeof(TRequest).Name); var sw = Stopwatch.StartNew(); var response = await next(request, ct); _logger.LogInformation("{Request} handled in {Elapsed}ms", typeof(TRequest).Name, sw.ElapsedMilliseconds); return response; } } ``` This applies **automatically** to every command, query, and request in your system. No manual wiring. No forgetting to add it to that new endpoint. 3. Consistent CQRS Semantics [#3-consistent-cqrs-semantics] The Mediator pattern naturally maps to CQRS. Commands mutate state. Queries read state. Notifications broadcast events. This vocabulary becomes a **shared language** across your team: * Seeing `ICommand` → you know it writes data * Seeing `IQuery` → you know it reads data * Seeing `INotification` → you know it's fire-and-forget This isn't just ceremony — it's **enforced architecture** that prevents common mistakes like commands that secretly read data or queries that mutate state. 4. Testability [#4-testability] Handlers are simple classes with a single `Handle` method. They're trivial to unit test in isolation — no HTTP context, no middleware chain, no controller setup: ```csharp [Fact] public async Task CreateOrder_ShouldReturnOrder() { var handler = new CreateOrderHandler(mockRepo); var result = await handler.Handle( new CreateOrderCommand("customer-1", items), CancellationToken.None); Assert.NotNull(result); } ``` Pipeline behaviors can also be tested independently, ensuring your cross-cutting concerns work correctly without spinning up the entire application. *** When Should You Use It? [#when-should-you-use-it] Strongly Recommended [#strongly-recommended] | Scenario | Why | | ----------------------------------------------------------------------------------- | -------------------------------------------------------- | | **Multiple entry points** (API + Workers + Jobs) | Shared pipeline and handlers eliminate duplication | | **Complex cross-cutting concerns** (auth, validation, caching, retry, transactions) | Pipeline behaviors apply them consistently | | **Large teams** working on the same codebase | Clear message contracts reduce coordination overhead | | **CQRS / Event-driven architectures** | Natural fit for command/query/event separation | | **Microservices** with shared business logic | Handlers are portable across service boundaries | | **Applications requiring auditability** | Pipeline behaviors can intercept and log every operation | Consider It [#consider-it] | Scenario | Why | | ------------------------------------------------- | ----------------------------------------------------------------------- | | **Medium-sized Web APIs** with growing complexity | Establishes patterns early before the codebase becomes hard to refactor | | **Applications with strict testing requirements** | Handler isolation makes unit testing straightforward | | **Teams adopting Clean/Hexagonal Architecture** | Mediator fits naturally as the application layer boundary | Probably Not Needed [#probably-not-needed] | Scenario | Why | | --------------------------------------- | --------------------------------------------------- | | **Small CRUD APIs** with 5-10 endpoints | The overhead of the pattern exceeds its benefits | | **Prototypes and MVPs** | Speed of development matters more than architecture | | **Simple scripts or utilities** | Direct method calls are simpler and sufficient | Even in smaller projects, TurboMediator's source-generated approach adds **near-zero runtime overhead** — so the cost of adopting it is primarily the learning curve, not performance. *** What About the "Service Class" Approach? [#what-about-the-service-class-approach] A common alternative is injecting service classes directly: ```csharp public class OrderController(IOrderService orderService) { public async Task Create(CreateOrderDto dto) => Ok(await orderService.CreateOrder(dto)); } ``` This works fine at small scale. But as your application grows: * **Service classes accumulate methods** and become god objects. * **Cross-cutting concerns are scattered** across services or duplicated. * **Adding a new concern** (e.g., caching) means touching every service method. * **Different entry points** (API vs. Worker) need to independently wire up the same logic. The Mediator pattern avoids all of these by giving you **one handler per operation** and a **single pipeline** for concerns. | | Service Classes | Mediator Pattern | | -------------------------- | -------------------------------------- | ------------------------------------ | | Adding validation | Modify each service method | Add one pipeline behavior | | Adding logging | Modify each service method | Add one pipeline behavior | | Adding caching | Modify each service method | Add one pipeline behavior | | New entry point (Worker) | Wire up services + recreate middleware | Just send the same message | | New team member onboarding | "Read these 15 service classes" | "Commands go here, queries go there" | *** The Real Cost of Not Using It [#the-real-cost-of-not-using-it] In large applications, the cost of **not** using a mediator becomes visible over time: 1. **Inconsistency** — Some operations have logging, some don't. Some have validation, some don't. New features ship without proper cross-cutting concerns because developers forget or don't know they exist. 2. **Duplication** — The same authorization check is written in three places. The same caching logic is copy-pasted. When a bug is found, only one copy gets fixed. 3. **Fragile refactoring** — Changing how a business operation works requires updating the controller, the worker, the job, and every test that sets up the full dependency chain. 4. **Difficult onboarding** — New developers have no clear pattern to follow. Every service does things slightly differently. Code reviews become debates about where logic should live. With a mediator, these problems are **structurally prevented**. The pipeline enforces consistency, handlers enforce single responsibility, and message contracts enforce clear boundaries. *** The Boilerplate Problem (Why I Built This) [#the-boilerplate-problem-why-i-built-this] Every time I worked on a real project using a mediator library, I ended up rewriting the same things from scratch: * Validation wired to FluentValidation * Retry and circuit breaker via Polly * Logging and tracing with OpenTelemetry and correlation IDs * EF Core unit-of-work transaction behavior * Per-user/per-tenant rate limiting * Test helpers to mock the mediator in unit tests None of it is hard. But it takes time, it drifts between projects, and you end up owning it forever. Every project had a slightly different version of the same `ValidationBehavior`, the same `LoggingBehavior`, the same test abstractions. That's what TurboMediator is trying to fix. Not just the mediator dispatch, but the entire surrounding ecosystem — built once, tested together, and ready to use. *** Why TurboMediator Specifically? [#why-turbomediator-specifically] If you're convinced the pattern is valuable, why choose TurboMediator over alternatives? * **Free. Always.** — MIT-licensed with no revenue thresholds, team size limits, or commercial restrictions. No license key to deploy. No pricing page to check before upgrading. * **Source-generated dispatch** — No reflection, no `Dictionary` lookups. A `switch` expression routes messages at compile time. * **Compile-time safety** — Forgot to create a handler? Missing a pipeline registration? The compiler tells you immediately — not a runtime exception in production. * **Native AOT ready** — No runtime code generation means full compatibility with `PublishAot`. * **20+ optional packages** — Validation, resilience, observability, persistence, sagas, rate limiting, feature flags — all built for the TurboMediator pipeline. * **Zero lock-in** — Your handlers are plain classes. Your messages are plain records. Migrating away (if you ever want to) means removing the `ICommand` interface and calling the handler directly. *** Summary [#summary] The Mediator pattern is not about adding indirection for the sake of it. It's about giving your application a **central nervous system** — a single pipeline through which all operations flow, where cross-cutting concerns are applied **once** and **consistently**, regardless of whether the trigger is an HTTP request, a queue message, or a scheduled job. If your application has more than one entry point, complex cross-cutting concerns, or a growing team — the Mediator pattern isn't just nice to have. It's a **force multiplier**. # Handlers import { Callout } from 'fumadocs-ui/components/callout'; Handlers contain the business logic that processes messages. Each message type has a corresponding handler interface. TurboMediator discovers handlers automatically at compile time via the source generator. Handler Interfaces [#handler-interfaces] | Handler Interface | Message Type | Return | | -------------------------------------- | --------------------- | ---------------------- | | `IRequestHandler` | `IRequest` | `ValueTask` | | `IRequestHandler` | `IRequest` | `ValueTask` | | `ICommandHandler` | `ICommand` | `ValueTask` | | `ICommandHandler` | `ICommand` | `ValueTask` | | `IQueryHandler` | `IQuery` | `ValueTask` | | `INotificationHandler` | `INotification` | `ValueTask` | Request Handlers [#request-handlers] ```csharp public record PingRequest : IRequest; public class PingHandler : IRequestHandler { public ValueTask Handle(PingRequest request, CancellationToken ct) { return new ValueTask("Pong!"); } } ``` For requests without a response: ```csharp public record LogEventRequest(string Message) : IRequest; public class LogEventHandler : IRequestHandler { private readonly ILogger _logger; public LogEventHandler(ILogger logger) => _logger = logger; public ValueTask Handle(LogEventRequest request, CancellationToken ct) { _logger.LogInformation("Event: {Message}", request.Message); return Unit.ValueTask; // or simply: return default; } } ``` Void message types (`IRequest`, `ICommand`) use the `Unit` type internally. In handlers you can return `Unit.ValueTask` (explicit) or `default` (shorthand) — both are equivalent. Command Handlers [#command-handlers] ```csharp public record CreateOrderCommand(string Product, int Quantity) : ICommand; public class CreateOrderHandler : ICommandHandler { private readonly IOrderRepository _repository; public CreateOrderHandler(IOrderRepository repository) { _repository = repository; } public async ValueTask Handle(CreateOrderCommand command, CancellationToken ct) { var order = new Order { Id = Guid.NewGuid(), Product = command.Product, Quantity = command.Quantity, CreatedAt = DateTime.UtcNow }; await _repository.AddAsync(order, ct); return order; } } ``` Query Handlers [#query-handlers] ```csharp public record GetOrderQuery(Guid Id) : IQuery; public class GetOrderHandler : IQueryHandler { private readonly IOrderRepository _repository; public GetOrderHandler(IOrderRepository repository) { _repository = repository; } public async ValueTask Handle(GetOrderQuery query, CancellationToken ct) { return await _repository.FindByIdAsync(query.Id, ct); } } ``` Notification Handlers [#notification-handlers] Notifications support **multiple handlers**. All registered handlers are invoked when a notification is published: ```csharp public record OrderPlacedNotification(Guid OrderId, string CustomerEmail) : INotification; // Handler 1: Send email public class OrderEmailHandler : INotificationHandler { private readonly IEmailService _emailService; public OrderEmailHandler(IEmailService emailService) => _emailService = emailService; public async ValueTask Handle(OrderPlacedNotification notification, CancellationToken ct) { await _emailService.SendOrderConfirmation(notification.CustomerEmail, notification.OrderId); } } // Handler 2: Update analytics public class OrderAnalyticsHandler : INotificationHandler { private readonly IAnalyticsService _analytics; public OrderAnalyticsHandler(IAnalyticsService analytics) => _analytics = analytics; public async ValueTask Handle(OrderPlacedNotification notification, CancellationToken ct) { await _analytics.TrackOrderPlaced(notification.OrderId); } } // Handler 3: Update inventory public class InventoryUpdateHandler : INotificationHandler { public ValueTask Handle(OrderPlacedNotification notification, CancellationToken ct) { Console.WriteLine($"Updating inventory for order {notification.OrderId}"); return default; } } ``` Dependency Injection [#dependency-injection] Handlers support constructor injection. All dependencies are resolved from the DI container: ```csharp public class CreateUserHandler : ICommandHandler { private readonly IUserRepository _repository; private readonly ILogger _logger; private readonly IMediator _mediator; public CreateUserHandler( IUserRepository repository, ILogger logger, IMediator mediator) { _repository = repository; _logger = logger; _mediator = mediator; } public async ValueTask Handle(CreateUserCommand command, CancellationToken ct) { _logger.LogInformation("Creating user {Name}", command.Name); var user = new User(Guid.NewGuid(), command.Name, command.Email); await _repository.AddAsync(user, ct); // Publish notification from within a handler await _mediator.Publish(new UserCreatedNotification(user.Id, user.Email), ct); return user; } } ``` Each message (except notifications) must have **exactly one** handler. The source generator will emit: * `TURBO001` warning if no handler is found * `TURBO002` error if multiple handlers exist for the same message * `TURBO008` error if the handler is not public Handler Lifetime [#handler-lifetime] By default, handlers are registered as **scoped** services — a new instance is created per request/scope. This is the safe default because handlers often inject scoped dependencies like `DbContext`, `ITenantContext`, or `HttpContext`. ```csharp // Default — Scoped (recommended for most apps) builder.Services.AddTurboMediator(); ``` You can change the lifetime if your handlers are stateless and don't inject scoped services: ```csharp // Singleton — reuses handler instances across all requests // ⚠️ Only safe when handlers DON'T inject scoped services (DbContext, ITenantContext, etc.) builder.Services.AddTurboMediator(ServiceLifetime.Singleton); ``` If handlers inject **scoped services** (e.g. EF Core `DbContext`, `ITenantContext`, `IHttpContextAccessor`, or any service registered with `AddScoped`), the lifetime **must** be `Scoped` (the default). Using `Singleton` will cause a runtime `InvalidOperationException`: *"Cannot consume scoped service from singleton"*. The `IMediator` instance itself is stateless — it resolves handlers on-demand from the DI container. The `lifetime` parameter controls the registration of **both** the mediator and all handlers. # Mediator Context import { Callout } from 'fumadocs-ui/components/callout'; `IMediatorContext` provides a way to flow **contextual data** through the entire pipeline — from pre-processors, through behaviors, to the handler and post-processors. IMediatorContext Interface [#imediatorcontext-interface] ```csharp public interface IMediatorContext { string CorrelationId { get; set; } string? CausationId { get; set; } string? UserId { get; set; } string? TenantId { get; set; } string? TraceId { get; set; } string? SpanId { get; set; } IDictionary Items { get; } // Keyed access void Set(string key, T value); T? Get(string key); bool TryGet(string key, out T? value); // Type-based access (uses type name as key) void Set(T value); T? Get(); bool TryGet(out T? value); } ``` Built-in Properties [#built-in-properties] | Property | Description | | --------------- | ------------------------------------------------- | | `CorrelationId` | Unique ID that groups related operations together | | `CausationId` | ID of the message that caused this message | | `UserId` | ID of the current user | | `TenantId` | ID of the current tenant (multi-tenancy) | | `TraceId` | Distributed tracing trace ID | | `SpanId` | Distributed tracing span ID | Using the Context [#using-the-context] Setting up [#setting-up] Register the mediator context with `WithMediatorContext()`: ```csharp builder.Services.AddTurboMediator(m => { m.WithMediatorContext(); }); ``` Accessing in handlers [#accessing-in-handlers] Inject `IMediatorContextAccessor` to access the current context: ```csharp public class CreateOrderHandler : ICommandHandler { private readonly IMediatorContextAccessor _contextAccessor; public CreateOrderHandler(IMediatorContextAccessor contextAccessor) { _contextAccessor = contextAccessor; } public async ValueTask Handle(CreateOrderCommand command, CancellationToken ct) { var context = _contextAccessor.Context; var userId = context?.UserId; var correlationId = context?.CorrelationId; // Use context data... return new Order { CreatedBy = userId }; } } ``` Storing custom data [#storing-custom-data] Use `Set(value)` and `Get()` for type-based access (uses the type name as the key), or `Set(key, value)` and `Get(key)` for explicit keys: ```csharp // Type-based access — convenient when you have one value per type context.Set(new RequestMetadata { IpAddress = "127.0.0.1" }); var metadata = context.Get(); // Keyed access — when you need multiple values of the same type context.Set("SourceIp", ipAddress); context.Set("TargetIp", targetAddress); var source = context.Get("SourceIp"); ``` Full behavior example: ```csharp // In a pre-processor or behavior public class EnrichContextBehavior : IPipelineBehavior where TMessage : IMessage { private readonly IMediatorContextAccessor _contextAccessor; private readonly IHttpContextAccessor _httpContextAccessor; public EnrichContextBehavior( IMediatorContextAccessor contextAccessor, IHttpContextAccessor httpContextAccessor) { _contextAccessor = contextAccessor; _httpContextAccessor = httpContextAccessor; } public async ValueTask Handle( TMessage message, MessageHandlerDelegate next, CancellationToken ct) { var context = _contextAccessor.Context!; // Set typed data context.Set("RequestMetadata", new RequestMetadata { IpAddress = _httpContextAccessor.HttpContext?.Connection.RemoteIpAddress?.ToString(), UserAgent = _httpContextAccessor.HttpContext?.Request.Headers.UserAgent.ToString() }); // Set keyed data context.Items["StartTime"] = DateTime.UtcNow; return await next(message, ct); } } ``` Child contexts [#child-contexts] `MediatorContext` (the default implementation) supports `CreateChild()` to spawn sub-operations that inherit parent context: ```csharp var childContext = context.CreateChild(); // Child inherits parent's CorrelationId but can have its own CausationId ``` IMediatorContextAccessor [#imediatorcontextaccessor] Uses `AsyncLocal` under the hood, ensuring context isolation across async flows: ```csharp public interface IMediatorContextAccessor { IMediatorContext? Context { get; set; } } ``` The mediator context is especially useful when combined with the Correlation ID, Structured Logging, and Telemetry behaviors from the Observability package — they all read from and write to this context. # Messages import { Callout } from 'fumadocs-ui/components/callout'; TurboMediator follows the **CQRS** (Command Query Responsibility Segregation) pattern with clear message type separation. All messages flow through the mediator pipeline and are routed to their respective handlers. Message Type Hierarchy [#message-type-hierarchy] ``` IMessage ├── IBaseRequest │ ├── IRequest // Generic request │ └── IRequest // Void request (returns Unit) ├── IBaseCommand │ ├── ICommand // Command with result │ └── ICommand // Void command (returns Unit) └── IBaseQuery └── IQuery // Query (always returns data) INotification // Fire-and-forget, multi-handler ``` Requests [#requests] Use `IRequest` for generic operations that don't fit neatly into Command or Query categories: ```csharp // Request with a response public record PingRequest : IRequest; // Request without response (returns Unit) public record LogEventRequest(string Message) : IRequest; ``` Commands [#commands] Use `ICommand` for **write operations** that modify state: ```csharp // Command with a response public record CreateUserCommand(string Name, string Email) : ICommand; // Command without response public record DeleteUserCommand(Guid UserId) : ICommand; // Command with result indicating success public record UpdateOrderCommand(Guid OrderId, string Status) : ICommand; ``` Queries [#queries] Use `IQuery` for **read operations** that return data without side effects: ```csharp public record GetUserQuery(Guid Id) : IQuery; public record GetAllUsersQuery : IQuery>; public record SearchProductsQuery(string Term, int Page) : IQuery>; ``` Notifications [#notifications] Use `INotification` for **events** that can have zero or more handlers: ```csharp public record UserCreatedNotification(Guid UserId, string Email) : INotification; public record OrderShippedNotification(Guid OrderId, DateTime ShippedAt) : INotification; ``` Unlike requests, commands, and queries, notifications can have **multiple handlers**. They are dispatched to all registered handlers based on the configured notification publisher strategy. The Unit Type [#the-unit-type] For messages that don't return a meaningful value, TurboMediator uses the `Unit` struct: ```csharp // IRequest (no generic parameter) is equivalent to IRequest public record DeleteUserCommand(Guid Id) : ICommand; // Handler returns Unit public class DeleteUserHandler : ICommandHandler { public ValueTask Handle(DeleteUserCommand command, CancellationToken ct) { // Perform deletion... return default; } } ``` `Unit` provides static helpers: * `Unit.Value` — The singleton `Unit` instance * `Unit.ValueTask` — A pre-completed `ValueTask` * `Unit.Task` — A pre-completed `Task` Sending Messages [#sending-messages] All messages are sent through the `IMediator` interface (or its sub-interfaces `ISender` and `IPublisher`): ```csharp public class MyService { private readonly IMediator _mediator; public MyService(IMediator mediator) => _mediator = mediator; public async Task DoWork() { // Send a query var user = await _mediator.Send(new GetUserQuery(someId)); // Send a command var newUser = await _mediator.Send(new CreateUserCommand("Alice", "alice@example.com")); // Send a request var pong = await _mediator.Send(new PingRequest()); // Publish a notification await _mediator.Publish(new UserCreatedNotification(newUser.Id, newUser.Email)); } } ``` Best Practices [#best-practices] 1. **Use records** — `record` types provide immutability, value equality, and concise syntax 2. **Follow CQRS** — Use `ICommand` for writes, `IQuery` for reads 3. **Keep messages simple** — Messages should be plain data containers, no behavior 4. **Use meaningful names** — Suffix with `Command`, `Query`, `Request`, or `Notification` 5. **Prefer specific types** — Use `ICommand`/`IQuery` over `IRequest` when the intent is clear # Notifications import { Callout } from 'fumadocs-ui/components/callout'; Notifications are messages that can be handled by **zero or more** handlers. They are ideal for decoupling side effects from the main business logic. Defining Notifications [#defining-notifications] ```csharp public record UserRegisteredNotification( Guid UserId, string Email, DateTime RegisteredAt ) : INotification; ``` Publishing Notifications [#publishing-notifications] Use `IPublisher` (or `IMediator`) to publish: ```csharp await mediator.Publish(new UserRegisteredNotification( user.Id, user.Email, DateTime.UtcNow )); ``` Notification Publishers [#notification-publishers] TurboMediator provides five built-in strategies for dispatching notifications to handlers: ForeachAwaitPublisher (Default) [#foreachawaitpublisher-default] Invokes handlers **sequentially**, one at a time. If a handler throws, subsequent handlers are not invoked. ```csharp builder.Services.AddTurboMediator(m => m.WithSequentialNotifications()); ``` TaskWhenAllPublisher [#taskwhenallpublisher] Invokes all handlers **in parallel** using `Task.WhenAll`. All handlers run concurrently. ```csharp builder.Services.AddTurboMediator(m => m.WithParallelNotifications()); ``` FireAndForgetPublisher [#fireandforgetpublisher] Dispatches handlers via `Task.Run` — **non-blocking**. The publish call returns immediately. ```csharp builder.Services.AddTurboMediator(m => m.WithFireAndForgetNotifications()); ``` Fire-and-forget notifications are not awaited. Exceptions in handlers will be lost unless you add your own error handling within the handlers. ContinueOnExceptionPublisher [#continueonexceptionpublisher] Invokes handlers **sequentially** but continues even if a handler throws. All exceptions are collected into an `AggregateException` thrown at the end. ```csharp builder.Services.AddTurboMediator(m => m.WithNotificationPublisher(ContinueOnExceptionPublisher.Instance)); ``` StopOnFirstExceptionPublisher [#stoponfirstexceptionpublisher] Invokes handlers **sequentially** and stops immediately on the first exception. ```csharp builder.Services.AddTurboMediator(m => m.WithNotificationPublisher(StopOnFirstExceptionPublisher.Instance)); ``` Custom Notification Publisher [#custom-notification-publisher] Implement `INotificationPublisher` to create your own strategy: ```csharp public class PriorityNotificationPublisher : INotificationPublisher { public async ValueTask Publish( INotificationHandler[] handlers, TNotification notification, CancellationToken cancellationToken) where TNotification : INotification { // Custom dispatch logic foreach (var handler in handlers) { await handler.Handle(notification, cancellationToken); } } } ``` Missing Handler Behavior [#missing-handler-behavior] By default, publishing a notification with no registered handlers is a no-op. You can change this: ```csharp builder.Services.AddTurboMediator(m => m.ThrowOnNoNotificationHandler()); ``` Practical Example [#practical-example] ```csharp // Define the notification public record PaymentProcessedNotification( Guid OrderId, decimal Amount, string Currency ) : INotification; // Handler 1: Send receipt public class SendReceiptHandler : INotificationHandler { private readonly IEmailService _email; public SendReceiptHandler(IEmailService email) => _email = email; public async ValueTask Handle(PaymentProcessedNotification n, CancellationToken ct) { await _email.SendReceiptAsync(n.OrderId, n.Amount, n.Currency); } } // Handler 2: Update accounting public class UpdateAccountingHandler : INotificationHandler { private readonly IAccountingService _accounting; public UpdateAccountingHandler(IAccountingService accounting) => _accounting = accounting; public async ValueTask Handle(PaymentProcessedNotification n, CancellationToken ct) { await _accounting.RecordPaymentAsync(n.OrderId, n.Amount, n.Currency); } } // Handler 3: Log for audit public class PaymentAuditHandler : INotificationHandler { private readonly ILogger _logger; public PaymentAuditHandler(ILogger logger) => _logger = logger; public ValueTask Handle(PaymentProcessedNotification n, CancellationToken ct) { _logger.LogInformation("Payment of {Amount} {Currency} processed for order {OrderId}", n.Amount, n.Currency, n.OrderId); return default; } } ``` # Streaming import { Callout } from 'fumadocs-ui/components/callout'; TurboMediator provides first-class support for **streaming responses** using `IAsyncEnumerable`. This is ideal for scenarios where you want to return data incrementally — large datasets, real-time feeds, or server-sent events. Stream Message Types [#stream-message-types] | Interface | Handler Interface | | --------------------------- | -------------------------------------------- | | `IStreamRequest` | `IStreamRequestHandler` | | `IStreamCommand` | `IStreamCommandHandler` | | `IStreamQuery` | `IStreamQueryHandler` | Defining Stream Messages [#defining-stream-messages] ```csharp // Stream request public record GenerateNumbersRequest(int Count) : IStreamRequest; // Stream query public record GetAllUsersQuery : IStreamQuery; // Stream command public record ProcessItemsCommand(IReadOnlyList Items) : IStreamCommand; ``` Stream Handlers [#stream-handlers] Stream handlers return `IAsyncEnumerable`: ```csharp public class GenerateNumbersHandler : IStreamRequestHandler { public async IAsyncEnumerable Handle( GenerateNumbersRequest request, [EnumeratorCancellation] CancellationToken ct) { for (int i = 1; i <= request.Count; i++) { ct.ThrowIfCancellationRequested(); await Task.Delay(100, ct); // Simulate work yield return i; } } } public class GetAllUsersHandler : IStreamQueryHandler { private readonly IUserRepository _repository; public GetAllUsersHandler(IUserRepository repository) { _repository = repository; } public async IAsyncEnumerable Handle( GetAllUsersQuery query, [EnumeratorCancellation] CancellationToken ct) { await foreach (var user in _repository.GetAllAsync(ct)) { yield return user; } } } ``` Consuming Streams [#consuming-streams] Use `CreateStream` on the mediator to get the `IAsyncEnumerable`: ```csharp // Consume a stream await foreach (var number in mediator.CreateStream(new GenerateNumbersRequest(10))) { Console.WriteLine($"Received: {number}"); } // Use with LINQ (System.Linq.Async) var users = await mediator .CreateStream(new GetAllUsersQuery()) .Where(u => u.IsActive) .Take(50) .ToListAsync(); ``` Stream Pipeline Behaviors [#stream-pipeline-behaviors] Streaming has dedicated pipeline interfaces: IStreamPipelineBehavior [#istreampipelinebehavior] Wraps the entire stream execution: ```csharp public class StreamLoggingBehavior : IStreamPipelineBehavior where TMessage : IStreamMessage { private readonly ILogger _logger; public StreamLoggingBehavior(ILogger> logger) { _logger = logger; } public async IAsyncEnumerable Handle( TMessage message, StreamHandlerDelegate next, [EnumeratorCancellation] CancellationToken ct) { _logger.LogInformation("Starting stream for {MessageType}", typeof(TMessage).Name); var count = 0; await foreach (var item in next()) { count++; yield return item; } _logger.LogInformation("Stream {MessageType} completed with {Count} items", typeof(TMessage).Name, count); } } ``` IStreamPreProcessor [#istreampreprocessor] Runs before the stream starts: ```csharp public class StreamValidationPreProcessor : IStreamPreProcessor where TMessage : IStreamMessage { public ValueTask Process(TMessage message, CancellationToken ct) { // Validate message before streaming starts return default; } } ``` IStreamPostProcessor [#istreampostprocessor] Wraps the stream after the handler — can transform, log, or add metrics to each item: ```csharp public class StreamMetricsPostProcessor : IStreamPostProcessor where TMessage : IStreamMessage { public async IAsyncEnumerable Process( TMessage message, IAsyncEnumerable stream, [EnumeratorCancellation] CancellationToken ct) { var count = 0; await foreach (var item in stream.WithCancellation(ct)) { count++; yield return item; } // Record metrics after stream completes } } ``` Registering Stream Behaviors [#registering-stream-behaviors] ```csharp builder.Services.AddTurboMediator(m => { m.WithStreamPipelineBehavior>(); m.WithStreamPreProcessor>(); m.WithStreamPostProcessor>(); }); ``` SSE Example (Minimal API) [#sse-example-minimal-api] ```csharp app.MapGet("/stream/numbers", async (HttpContext context, IMediator mediator) => { context.Response.ContentType = "text/event-stream"; await foreach (var number in mediator.CreateStream(new GenerateNumbersRequest(100))) { await context.Response.WriteAsync($"data: {number}\n\n"); await context.Response.Body.FlushAsync(); } }); ``` Stream pipeline behaviors use separate interfaces (`IStreamPipelineBehavior`, `IStreamPreProcessor`, `IStreamPostProcessor`) that are independent from the regular pipeline behaviors. # Authorization import { Callout } from 'fumadocs-ui/components/callout'; The authorization behavior provides attribute-based access control for mediator operations, checking roles and policies before the handler executes. Installation [#installation] ```bash dotnet add package TurboMediator.Enterprise ``` Using Attributes [#using-attributes] [Authorize] [#authorize] ```csharp // Require authentication [Authorize] public record GetProfileQuery : IQuery; // Require specific role [Authorize(Roles = "Admin")] public record DeleteUserCommand(Guid UserId) : ICommand; // Require a policy [Authorize(Policy = "CanManageOrders")] public record CancelOrderCommand(Guid OrderId) : ICommand; // Multiple requirements [Authorize(Roles = "Admin,Manager", Policy = "CanEditProducts")] public record UpdateProductCommand(Guid ProductId, string Name, decimal Price) : ICommand; // Authentication scheme [Authorize(AuthenticationSchemes = "Bearer")] public record GetApiDataQuery : IQuery; ``` [AllowAnonymous] [#allowanonymous] Override `[Authorize]` on specific operations: ```csharp [AllowAnonymous] public record GetPublicProductsQuery : IQuery>; ``` IUserContext [#iusercontext] Implement `IUserContext` to provide user information: ```csharp public interface IUserContext { ClaimsPrincipal? User { get; } bool IsAuthenticated { get; } } // ASP.NET Core implementation public class HttpUserContext : IUserContext { private readonly IHttpContextAccessor _httpContextAccessor; public HttpUserContext(IHttpContextAccessor httpContextAccessor) { _httpContextAccessor = httpContextAccessor; } public ClaimsPrincipal? User => _httpContextAccessor.HttpContext?.User; public bool IsAuthenticated => User?.Identity?.IsAuthenticated ?? false; } ``` IAuthorizationPolicyProvider [#iauthorizationpolicyprovider] Implement custom policy evaluation: ```csharp public class CustomPolicyProvider : IAuthorizationPolicyProvider { public ValueTask EvaluatePolicyAsync( ClaimsPrincipal user, string policy, CancellationToken ct) { return policy switch { "CanManageOrders" => new ValueTask( user.HasClaim("permission", "orders:manage")), "CanEditProducts" => new ValueTask( user.IsInRole("Admin") || user.HasClaim("permission", "products:edit")), _ => new ValueTask(false) }; } } ``` Registration [#registration] ```csharp builder.Services.AddTurboMediator(m => { m.WithAuthorization(); m.WithUserContext(); m.WithAuthorizationPolicies(policies => { policies.AddPolicy("CanManageOrders", u => u.HasClaim("permission", "orders:manage")); policies.AddPolicy("CanEditProducts", u => u.IsInRole("Admin") || u.HasClaim("permission", "products:edit")); }); }); ``` UnauthorizedException [#unauthorizedexception] Thrown when authorization fails: ```csharp try { await mediator.Send(new DeleteUserCommand(userId)); } catch (UnauthorizedException ex) { // ex.MessageType — the message type that was denied // ex.Policy — the policy that failed // ex.RequiredRoles — required roles that the user doesn't have return Results.Forbid(); } ``` Practical Example [#practical-example] ```csharp // Define messages with authorization [Authorize(Roles = "Admin")] public record PromoteUserCommand(Guid UserId, string NewRole) : ICommand; [Authorize(Policy = "CanViewReports")] public record GetSalesReportQuery(DateRange Range) : IQuery; [AllowAnonymous] public record GetPublicStatsQuery : IQuery; // Register builder.Services.AddHttpContextAccessor(); builder.Services.AddTurboMediator(m => { m.WithAuthorization(); m.WithUserContext(); }); // Usage in API app.MapDelete("/users/{id}", async (Guid id, IMediator mediator) => { try { await mediator.Send(new PromoteUserCommand(id, "Manager")); return Results.Ok(); } catch (UnauthorizedException) { return Results.Forbid(); } }); ``` The authorization behavior checks `[AllowAnonymous]` first. If present, authorization is skipped entirely. Otherwise, it validates authentication, roles, and policies in order. # Deduplication import { Callout } from 'fumadocs-ui/components/callout'; The deduplication behavior prevents the same message from being processed more than once by using idempotency keys and a pluggable store. IIdempotentMessage [#iidempotentmessage] Mark messages that support deduplication: ```csharp public interface IIdempotentMessage { string IdempotencyKey { get; } } // Example public record ProcessPaymentCommand( Guid PaymentId, decimal Amount, string Currency ) : ICommand, IIdempotentMessage { public string IdempotencyKey => $"payment:{PaymentId}"; } ``` IIdempotencyStore [#iidempotencystore] ```csharp public interface IIdempotencyStore { ValueTask TryAcquireAsync(string key, TimeSpan ttl, CancellationToken ct); ValueTask?> GetAsync(string key, CancellationToken ct); ValueTask SetAsync(string key, T response, TimeSpan ttl, CancellationToken ct); ValueTask ReleaseAsync(string key, CancellationToken ct); } ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithDeduplication(options => { options.TimeToLive = TimeSpan.FromHours(24); options.ThrowOnDuplicate = false; options.MaxWaitRetries = 5; options.WaitRetryInterval = TimeSpan.FromMilliseconds(200); options.ReleaseOnError = true; }); m.WithInMemoryIdempotencyStore(); }); ``` DeduplicationOptions [#deduplicationoptions] | Option | Default | Description | | ------------------- | -------- | ------------------------------------------------------ | | `TimeToLive` | 24 hours | How long idempotency records are kept | | `ThrowOnDuplicate` | `false` | Throw exception on duplicate vs return cached response | | `MaxWaitRetries` | 5 | Retries when a concurrent duplicate is being processed | | `WaitRetryInterval` | 200ms | Delay between retries | | `ReleaseOnError` | `true` | Release lock if handler throws | How It Works [#how-it-works] 1. Checks if `IdempotencyKey` exists in the store 2. If found → return cached response (or throw if `ThrowOnDuplicate`) 3. If not found → acquire lock and execute handler 4. Store the response with the key 5. If handler throws and `ReleaseOnError` is true → release the lock For concurrent duplicate requests: 1. First request acquires the lock and processes 2. Second request waits (up to `MaxWaitRetries`) for the first to complete 3. Returns the cached response once available Practical Example [#practical-example] ```csharp public record ChargeCustomerCommand( string PaymentIntentId, decimal Amount, string CustomerId ) : ICommand, IIdempotentMessage { public string IdempotencyKey => $"charge:{PaymentIntentId}"; } public class ChargeCustomerHandler : ICommandHandler { private readonly IPaymentGateway _gateway; public ChargeCustomerHandler(IPaymentGateway gateway) => _gateway = gateway; public async ValueTask Handle( ChargeCustomerCommand command, CancellationToken ct) { // Safe to execute — deduplication ensures this runs only once // per PaymentIntentId, even with network retries return await _gateway.ChargeAsync( command.CustomerId, command.Amount, ct); } } ``` Deduplication is essential for webhook handlers, payment processing, and any operation where retries from external systems could cause duplicates. # Multi-Tenancy import { Callout } from 'fumadocs-ui/components/callout'; The multi-tenancy behavior ensures tenant context is present and validated for operations that require it. Core Interfaces [#core-interfaces] ITenantContext [#itenantcontext] Provides current tenant information: ```csharp public interface ITenantContext { string? TenantId { get; } bool HasTenant { get; } string? TenantName { get; } } ``` ITenantAware [#itenantaware] Mark messages that belong to a specific tenant: ```csharp public interface ITenantAware { string? TenantId { get; } } ``` Using the Attribute [#using-the-attribute] ```csharp [RequiresTenant] public record GetTenantDataQuery(string DataKey) : IQuery; ``` Tenant-Aware Messages [#tenant-aware-messages] ```csharp public record CreateTenantDocumentCommand( string TenantId, string Title, string Content ) : ICommand, ITenantAware; ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithMultiTenancy(); m.WithTenantContext(); }); ``` Implementing ITenantContext [#implementing-itenantcontext] ```csharp public class HttpTenantContext : ITenantContext { private readonly IHttpContextAccessor _httpContextAccessor; public HttpTenantContext(IHttpContextAccessor httpContextAccessor) { _httpContextAccessor = httpContextAccessor; } public string? TenantId => _httpContextAccessor.HttpContext?.Request.Headers["X-Tenant-Id"].FirstOrDefault() ?? _httpContextAccessor.HttpContext?.User?.FindFirst("tenant_id")?.Value; public bool HasTenant => !string.IsNullOrEmpty(TenantId); public string? TenantName => _httpContextAccessor.HttpContext?.User?.FindFirst("tenant_name")?.Value; } ``` Validation Rules [#validation-rules] The `TenantBehavior` validates: 1. If `[RequiresTenant]` is present, `ITenantContext.HasTenant` must be `true` 2. If the message implements `ITenantAware`, the message's `TenantId` must match `ITenantContext.TenantId` 3. Throws an exception if validation fails Practical Example [#practical-example] ```csharp // Tenant-isolated queries [RequiresTenant] public record GetTenantUsersQuery : IQuery>; public class GetTenantUsersHandler : IQueryHandler> { private readonly ITenantContext _tenantContext; private readonly IUserRepository _repository; public GetTenantUsersHandler(ITenantContext tenantContext, IUserRepository repository) { _tenantContext = tenantContext; _repository = repository; } public async ValueTask> Handle( GetTenantUsersQuery query, CancellationToken ct) { // TenantId is guaranteed to be present by the behavior return await _repository.GetByTenantAsync(_tenantContext.TenantId!, ct); } } ``` The tenant behavior should be registered **before** the handler in the pipeline to ensure tenant validation happens early. Register it after authentication/authorization. # Installation Requirements [#requirements] * .NET 8.0+ (or .NET Standard 2.0 for the abstractions) * C# 11+ (for source generator support) Core Packages [#core-packages] Every TurboMediator project requires these two packages: ```bash title="Terminal" dotnet add package TurboMediator.Abstractions dotnet add package TurboMediator.SourceGenerator ``` The **Abstractions** package contains all core interfaces (`IRequest`, `ICommand`, `IQuery`, `INotification`, handlers, pipeline behaviors, etc.). The **SourceGenerator** package contains the Roslyn incremental source generator that produces the `Mediator` implementation class at compile time. Optional Packages [#optional-packages] Install only the packages you need. Each package is independent and adds specific pipeline behaviors: ```bash title="Validation" dotnet add package TurboMediator.FluentValidation ``` ```bash title="Resilience (Retry, Circuit Breaker, Timeout, Fallback, Hedging)" dotnet add package TurboMediator.Resilience ``` ```bash title="Result Pattern (Functional Error Handling)" dotnet add package TurboMediator.Result ``` ```bash title="Observability (Telemetry, Metrics, Correlation, Logging, Health Checks)" dotnet add package TurboMediator.Observability ``` ```bash title="Caching (In-Memory, Custom Providers)" dotnet add package TurboMediator.Caching ``` ```bash title="Caching - Redis Provider" dotnet add package TurboMediator.Caching.Redis ``` ```bash title="Validation (Built-in Lightweight Validator)" dotnet add package TurboMediator.Validation ``` ```bash title="Enterprise (Authorization, Multi-Tenancy, Deduplication)" dotnet add package TurboMediator.Enterprise ``` ```bash title="Scheduling (Cron Jobs, Recurring Jobs)" dotnet add package TurboMediator.Scheduling dotnet add package TurboMediator.Scheduling.EntityFramework ``` ```bash title="Rate Limiting & Bulkhead" dotnet add package TurboMediator.RateLimiting ``` ```bash title="Distributed Locking" dotnet add package TurboMediator.DistributedLocking dotnet add package TurboMediator.DistributedLocking.Redis ``` ```bash title="Persistence (Transactions, Outbox, Audit)" dotnet add package TurboMediator.Persistence dotnet add package TurboMediator.Persistence.EntityFramework ``` ```bash title="Saga" dotnet add package TurboMediator.Saga dotnet add package TurboMediator.Saga.EntityFramework ``` ```bash title="State Machine" dotnet add package TurboMediator.StateMachine dotnet add package TurboMediator.StateMachine.EntityFramework ``` ```bash title="Feature Flags" dotnet add package TurboMediator.FeatureFlags dotnet add package TurboMediator.FeatureFlags.FeatureManagement ``` ```bash title="Batching" dotnet add package TurboMediator.Batching ``` ```bash title="Testing" dotnet add package TurboMediator.Testing ``` ```bash title="CLI Tool (global)" dotnet tool install --global TurboMediator.Cli ``` Project Setup [#project-setup] After installing the packages, register TurboMediator in your DI container: ```csharp title="Program.cs" using Microsoft.Extensions.DependencyInjection; var builder = WebApplication.CreateBuilder(args); // Basic registration - auto-discovers all handlers builder.Services.AddTurboMediator(); // Or with configuration builder.Services.AddTurboMediator(mediator => { mediator .WithSequentialNotifications(); }); var app = builder.Build(); ``` The source generator automatically generates the `AddTurboMediator()` extension method with all discovered handler registrations. # Quick Start import { Steps, Step } from 'fumadocs-ui/components/steps'; import { Callout } from 'fumadocs-ui/components/callout'; This guide walks you through building a simple application with TurboMediator. Install the packages [#install-the-packages] ```bash title="Terminal" dotnet add package TurboMediator.Abstractions dotnet add package TurboMediator.SourceGenerator ``` Define your messages [#define-your-messages] TurboMediator supports four message types following CQRS principles: ```csharp title="Messages.cs" using TurboMediator; // Query — read operation, returns data public record GetUserQuery(Guid Id) : IQuery; // Command — write operation, returns result public record CreateUserCommand(string Name, string Email) : ICommand; // Request — generic request with response public record PingRequest : IRequest; // Notification — fire-and-forget, multiple handlers public record UserCreatedNotification(Guid UserId, string Email) : INotification; // Supporting types public record User(Guid Id, string Name, string Email); ``` Create handlers [#create-handlers] Each message type has a corresponding handler interface: ```csharp title="Handlers.cs" using TurboMediator; public class GetUserHandler : IQueryHandler { public ValueTask Handle(GetUserQuery query, CancellationToken ct) { var user = new User(query.Id, "John Doe", "john@example.com"); return new ValueTask(user); } } public class CreateUserHandler : ICommandHandler { public ValueTask Handle(CreateUserCommand command, CancellationToken ct) { var user = new User(Guid.NewGuid(), command.Name, command.Email); return new ValueTask(user); } } public class PingHandler : IRequestHandler { public ValueTask Handle(PingRequest request, CancellationToken ct) { return new ValueTask("Pong!"); } } // Multiple notification handlers are allowed public class EmailNotificationHandler : INotificationHandler { public ValueTask Handle(UserCreatedNotification notification, CancellationToken ct) { Console.WriteLine($"Sending welcome email to {notification.Email}"); return default; } } public class LogNotificationHandler : INotificationHandler { public ValueTask Handle(UserCreatedNotification notification, CancellationToken ct) { Console.WriteLine($"User created: {notification.UserId}"); return default; } } ``` Register services [#register-services] ```csharp title="Program.cs" using Microsoft.Extensions.DependencyInjection; var builder = WebApplication.CreateBuilder(args); // Register TurboMediator — handlers are auto-discovered builder.Services.AddTurboMediator(); var app = builder.Build(); ``` Use the mediator [#use-the-mediator] Inject `IMediator` (or `ISender` / `IPublisher`) and send messages: ```csharp title="Usage" app.MapGet("/ping", async (IMediator mediator) => { var result = await mediator.Send(new PingRequest()); return Results.Ok(result); // "Pong!" }); app.MapGet("/users/{id}", async (Guid id, IMediator mediator) => { var user = await mediator.Send(new GetUserQuery(id)); return user is not null ? Results.Ok(user) : Results.NotFound(); }); app.MapPost("/users", async (CreateUserCommand command, IMediator mediator) => { var user = await mediator.Send(command); // Publish notification — all handlers will be invoked await mediator.Publish(new UserCreatedNotification(user.Id, user.Email)); return Results.Created($"/users/{user.Id}", user); }); app.Run(); ``` The source generator automatically discovers all handlers in your project and generates compile-time dispatch code. You don't need to manually register each handler. Compile-Time Safety [#compile-time-safety] TurboMediator validates your handlers at build time. You'll get compiler errors for: | Diagnostic | Description | | ---------- | ------------------------------------------------------------- | | `TURBO001` | Missing handler for a message | | `TURBO002` | Multiple handlers for the same message (except notifications) | | `TURBO003` | Invalid handler signature | | `TURBO004` | Response type mismatch | | `TURBO008` | Handler must be public | What's Next? [#whats-next] Now that you have a working setup, explore the features: * [Messages & Handlers](/docs/core/messages) — Learn about all message types * [Pipeline Behaviors](/docs/pipeline/behaviors) — Add cross-cutting concerns * [Streaming](/docs/core/streaming) — Use `IAsyncEnumerable` for streaming * [Notifications](/docs/core/notifications) — Multi-handler publish strategies # Source Generator import { Callout } from 'fumadocs-ui/components/callout'; TurboMediator uses a **Roslyn Incremental Source Generator** to produce compile-time optimized code. This is the core of the library's performance advantage. What Gets Generated [#what-gets-generated] The source generator produces: 1. **`Mediator` class** — The concrete `IMediator` implementation with `switch`-expression dispatch for each message type 2. **`ExecuteWithPipeline` method** — Resolves pre-processors, pipeline behaviors, exception handlers, and post-processors from DI and chains them 3. **`AddTurboMediator()` extension** — DI registration that wires all discovered handlers Example Generated Code [#example-generated-code] For a project with `PingRequest` and `GetUserQuery`: ```csharp // Generated by TurboMediator Source Generator internal sealed class Mediator : IMediator { private readonly IServiceProvider _serviceProvider; public Mediator(IServiceProvider serviceProvider) { _serviceProvider = serviceProvider; } public ValueTask Send(IRequest request, CancellationToken ct) { return request switch { PingRequest msg => ExecuteWithPipeline(msg, ct), _ => throw new InvalidOperationException($"No handler for {request.GetType().Name}") }; } public ValueTask Send(IQuery query, CancellationToken ct) { return query switch { GetUserQuery msg => ExecuteWithPipeline(msg, ct), _ => throw new InvalidOperationException($"No handler for {query.GetType().Name}") }; } // ... similar for Commands and Notifications } ``` No Reflection [#no-reflection] The key benefit is that dispatch is a **compile-time `switch` expression** — no `Dictionary` lookups, no `Activator.CreateInstance`, no `MethodInfo.Invoke`. This makes it: * Faster than reflection-based mediators * Compatible with **Native AOT** publishing * Fully **trimmable** by the .NET linker Build-Time Diagnostics [#build-time-diagnostics] The source generator emits diagnostics during compilation: | Code | Severity | Description | | ---------- | -------- | -------------------------------------------------- | | `TURBO001` | Warning | No handler found for message type | | `TURBO002` | Error | Multiple handlers for the same message type | | `TURBO003` | Error | Invalid handler signature | | `TURBO004` | Error | Response type mismatch between message and handler | | `TURBO005` | Warning | Handler class is abstract (cannot be instantiated) | | `TURBO006` | Warning | No stream handler found for stream message | | `TURBO007` | Error | Multiple stream handlers for the same message | | `TURBO008` | Error | Handler class must be public | | `TURBO009` | Info | Duplicate notification handler detected | Example Build Errors [#example-build-errors] ``` error TURBO002: Multiple handlers found for 'GetUserQuery': GetUserHandler, AnotherGetUserHandler error TURBO008: Handler 'InternalHandler' must be public to be discovered by TurboMediator warning TURBO001: No handler found for message 'OrphanedCommand' ``` How It Works Internally [#how-it-works-internally] 1. **Parser Phase**: Scans the syntax tree for: * Classes implementing handler interfaces (`IRequestHandler`, `ICommandHandler`, `IQueryHandler`, `INotificationHandler`, streaming variants) * Types implementing message interfaces (`IRequest`, `ICommand`, `IQuery`, `INotification`, streaming variants) 2. **Validation Phase**: Checks for: * Missing handlers * Duplicate handlers * Signature mismatches * Accessibility (public) 3. **Emission Phase**: Generates: * `Mediator` class with switch-expression routing * Pipeline execution with DI-resolved behaviors * `AddTurboMediator()` service collection extension # Correlation ID The correlation ID behavior ensures every operation has a unique identifier that flows through the entire pipeline and can be propagated to downstream services. Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithCorrelationId(options => { options.HeaderName = "X-Correlation-ID"; options.GenerateIfMissing = true; options.AddToActivityBaggage = true; options.AddToLogScope = true; options.PropagateToHttpClient = true; }); m.WithMediatorContext(); // Required for context storage }); ``` CorrelationOptions [#correlationoptions] | Option | Default | Description | | ------------------------ | -------------------- | ------------------------------------ | | `HeaderName` | `"X-Correlation-ID"` | HTTP header name | | `GenerateIfMissing` | `true` | Auto-generate if not provided | | `CorrelationIdGenerator` | `Guid.NewGuid()` | Custom ID generator function | | `AddToActivityBaggage` | `true` | Add to `Activity.Current.Baggage` | | `AddToLogScope` | `true` | Add to `ILogger` scope | | `PropagateToHttpClient` | `true` | Add to outgoing HTTP headers | | `CorrelationIdProvider` | `null` | Custom provider to read existing IDs | Reading the Correlation ID [#reading-the-correlation-id] Access through `IMediatorContextAccessor`: ```csharp public class MyHandler : ICommandHandler { private readonly IMediatorContextAccessor _contextAccessor; private readonly ILogger _logger; public MyHandler( IMediatorContextAccessor contextAccessor, ILogger logger) { _contextAccessor = contextAccessor; _logger = logger; } public async ValueTask Handle(MyCommand command, CancellationToken ct) { var correlationId = _contextAccessor.Context?.CorrelationId; _logger.LogInformation("Processing with correlation ID: {CorrelationId}", correlationId); // The correlation ID is automatically included in log scope // and Activity baggage return new MyResult(); } } ``` Propagation to Downstream Services [#propagation-to-downstream-services] When `PropagateToHttpClient` is enabled, the correlation ID is automatically added to outgoing HTTP requests via the configured header: ```csharp // The correlation ID flows automatically to downstream HTTP calls public class OrderHandler : ICommandHandler { private readonly HttpClient _httpClient; public OrderHandler(HttpClient httpClient) => _httpClient = httpClient; public async ValueTask Handle(CreateOrderCommand cmd, CancellationToken ct) { // X-Correlation-ID header is automatically added var inventory = await _httpClient.GetFromJsonAsync( $"/api/inventory/{cmd.ProductId}", ct); return new Order { /* ... */ }; } } ``` # Health Checks TurboMediator integrates with ASP.NET Core health checks to monitor the health of the mediator system. Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithHealthCheck(options => { options.Tags = new[] { "ready" }; options.CheckHandlerRegistration = true; options.CheckCircuitBreakers = true; options.CheckSagaStore = true; options.CheckOutboxBacklog = true; options.MaxOutboxBacklog = 1000; options.Timeout = TimeSpan.FromSeconds(5); }); }); // Map health endpoint app.MapHealthChecks("/health"); ``` TurboMediatorHealthCheckOptions [#turbomediatorhealthcheckoptions] | Option | Default | Description | | -------------------------- | ---------------------------- | ------------------------------------------------- | | `CheckHandlerRegistration` | `true` | Verify all message types have registered handlers | | `CheckCircuitBreakers` | `true` | Report unhealthy if any circuit breaker is open | | `CheckSagaStore` | `true` | Check saga store connectivity | | `CheckOutboxBacklog` | `true` | Check outbox backlog size | | `MaxOutboxBacklog` | `1000` | Maximum outbox backlog before reporting degraded | | `DegradedThreshold` | `0.8` | Percentage of MaxOutboxBacklog before degraded | | `Timeout` | `5s` | Timeout for health check operations | | `Tags` | `["turbomediator", "ready"]` | Tags for filtering health checks | | `IncludeDetails` | `true` | Include detailed information in response | Health Check Result [#health-check-result] The health check verifies: * Mediator is registered and can be resolved from DI * Handler resolution works correctly * Pipeline is configured properly Returns `Healthy`, `Degraded`, or `Unhealthy` based on the check results. # Metrics The metrics behavior provides detailed request-level metrics using `System.Diagnostics.Metrics`, independent of the telemetry behavior. Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithMetrics(options => { options.MeterName = "MyApp.Mediator"; options.MeterVersion = "1.0.0"; }); }); ``` MetricsOptions [#metricsoptions] | Option | Default | Description | | --------------------------- | ----------------- | ---------------------------------------- | | `EnableLatencyHistogram` | `true` | Record request duration histogram | | `EnableThroughputCounter` | `true` | Count total requests | | `EnableErrorCounter` | `true` | Count errors | | `EnableInFlightGauge` | `true` | Track in-flight requests | | `LatencyBuckets` | Default | Custom histogram buckets | | `CustomLabels` | `[]` | Additional label names (`IList`) | | `CustomLabelValuesProvider` | `null` | Provider for custom label values | | `MeterName` | `"TurboMediator"` | Meter name for filtering | | `MeterVersion` | `"1.0.0"` | Meter version | Instruments Created [#instruments-created] | Instrument | Type | Description | | -------------------------------- | ------------- | ------------------------------- | | `turbomediator.handler.duration` | Histogram | Handler execution duration (ms) | | `turbomediator.handler.requests` | Counter | Total handler invocations | | `turbomediator.handler.errors` | Counter | Handler errors | | `turbomediator.handler.inflight` | UpDownCounter | Currently executing handlers | All metrics include `message_type` as a label. Practical Example [#practical-example] ```csharp builder.Services.AddTurboMediator(m => { m.WithMetrics(options => { options.LatencyBuckets = new double[] { 5, 10, 25, 50, 100, 250, 500, 1000 }; options.CustomLabels = new List { "service", "environment" }; options.CustomLabelValuesProvider = () => new Dictionary { ["service"] = "order-service", ["environment"] = builder.Environment.EnvironmentName }; }); }); // Export to Prometheus builder.Services.AddOpenTelemetry() .WithMetrics(metrics => { metrics .AddMeter("TurboMediator") .AddPrometheusExporter(); }); ``` # Structured Logging The structured logging behavior provides detailed, structured logs for every mediator operation with support for sensitive property masking and slow operation detection. Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithStructuredLogging(options => { options.IncludeMessageType = true; options.IncludeHandlerName = true; options.IncludeDuration = true; options.IncludeCorrelationId = true; options.IncludeMessageProperties = true; options.IncludeResponse = false; options.SlowOperationThreshold = TimeSpan.FromSeconds(1); options.SensitivePropertyNames = new HashSet { "Password", "CreditCard", "SSN", "Token" }; }); }); ``` StructuredLoggingOptions [#structuredloggingoptions] | Option | Default | Description | | -------------------------- | -------------------------------------------------------------------------------------------------------------- | ---------------------------------------------- | | `IncludeMessageType` | `true` | Log the message type name | | `IncludeHandlerName` | `true` | Log the handler class name | | `IncludeDuration` | `true` | Log execution duration | | `IncludeCorrelationId` | `true` | Log correlation ID | | `IncludeMessageProperties` | `false` | Log message property values | | `IncludeResponse` | `false` | Log response value | | `SensitivePropertyNames` | Password, Secret, Token, ApiKey, ConnectionString, CreditCard, CardNumber, Cvv, Pin, Ssn, SocialSecurityNumber | Properties to mask in logs | | `SuccessLogLevel` | `Information` | Log level for successes | | `ErrorLogLevel` | `Error` | Log level for errors | | `SlowOperationLogLevel` | `Warning` | Log level for slow operations | | `SlowOperationThreshold` | 1 second | Duration to be considered slow | | `ShouldLog` | `null` | Predicate to skip logging for certain messages | Sensitive Property Masking [#sensitive-property-masking] Properties listed in `SensitivePropertyNames` are replaced with `***REDACTED***` in log output: ```csharp // Configuration options.SensitivePropertyNames = new HashSet { "Password", "Token" }; // Message public record LoginCommand(string Username, string Password) : ICommand; // Log output: // Handling LoginCommand: { Username = "alice", Password = "***REDACTED***" } ``` Slow Operation Detection [#slow-operation-detection] Operations exceeding `SlowOperationThreshold` are logged at a higher level: ``` [Warning] Slow handler detected: GenerateReportQuery took 2345ms (threshold: 1000ms) ``` Log Output Examples [#log-output-examples] ``` [Information] Handling CreateUserCommand: MessageType: CreateUserCommand CorrelationId: abc-123 Properties: { Name = "Alice", Email = "alice@example.com" } [Information] Handled CreateUserCommand in 45ms HandlerName: CreateUserHandler Duration: 45ms [Error] Error handling CreateUserCommand after 12ms HandlerName: CreateUserHandler Exception: System.InvalidOperationException: Email already exists ``` Conditional Logging [#conditional-logging] Skip logging for specific message types: ```csharp options.ShouldLog = (messageType) => { // Don't log health check queries return messageType != typeof(HealthCheckQuery); }; ``` # Telemetry import { Callout } from 'fumadocs-ui/components/callout'; The telemetry behavior integrates with OpenTelemetry to provide distributed tracing and metrics for all mediator operations. Installation [#installation] ```bash dotnet add package TurboMediator.Observability ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { // Apply to all messages m.WithTelemetry(options => { options.RecordTraces = true; options.RecordMetrics = true; options.RecordExceptionStackTrace = false; }); // Or apply to a specific message type m.WithTelemetry(options => { options.RecordExceptionStackTrace = true; }); }); ``` TelemetryOptions [#telemetryoptions] | Option | Default | Description | | --------------------------- | ------- | ---------------------------------------- | | `RecordTraces` | `true` | Create Activity spans for each operation | | `RecordMetrics` | `true` | Record counters and histograms | | `RecordExceptionStackTrace` | `false` | Include stack traces in error events | Built-in Metrics [#built-in-metrics] `TurboMediatorTelemetry` provides these static instruments: | Metric | Type | Description | | ----------------------------------- | --------- | -------------------------------- | | `turbomediator.requests.total` | Counter | Total requests processed | | `turbomediator.requests.success` | Counter | Successful requests | | `turbomediator.requests.failure` | Counter | Failed requests | | `turbomediator.requests.duration` | Histogram | Request duration in milliseconds | | `turbomediator.notifications.total` | Counter | Total notifications published | Custom Telemetry Tags [#custom-telemetry-tags] Implement `ITelemetryEnriched` on your messages to add custom tags to traces: ```csharp public record ProcessOrderCommand(Guid OrderId, string Region) : ICommand, ITelemetryEnriched { public IEnumerable> GetTelemetryTags() { return new Dictionary { ["order.id"] = OrderId.ToString(), ["order.region"] = Region }; } } ``` OpenTelemetry Integration [#opentelemetry-integration] Register the TurboMediator source with your OpenTelemetry configuration: ```csharp builder.Services.AddOpenTelemetry() .WithTracing(tracing => { tracing .AddSource(TurboMediatorTelemetry.ActivitySource.Name) .AddAspNetCoreInstrumentation() .AddOtlpExporter(); }) .WithMetrics(metrics => { metrics .AddMeter(TurboMediatorTelemetry.Meter.Name) .AddOtlpExporter(); }); ``` Activity Spans [#activity-spans] Each mediator operation creates an `Activity` span with: * **Name**: Message type name (e.g., `CreateUserCommand`) * **Tags**: Message type, handler type, success/failure status, duration * **Events**: Exception details on failure * **Custom tags**: From `ITelemetryEnriched` implementations The telemetry behavior automatically correlates with ASP.NET Core and HttpClient traces when using OpenTelemetry, giving you end-to-end distributed tracing. # Batching import { Callout } from 'fumadocs-ui/components/callout'; The batching behavior collects multiple individual queries and processes them together in a single batch, significantly reducing database round-trips and improving throughput. Installation [#installation] ```bash dotnet add package TurboMediator.Batching ``` Defining Batchable Queries [#defining-batchable-queries] Mark queries with `IBatchableQuery`: ```csharp public record GetProductByIdQuery(Guid ProductId) : IBatchableQuery; public record GetUserByIdQuery(Guid UserId) : IBatchableQuery; ``` Creating Batch Handlers [#creating-batch-handlers] Implement `IBatchHandler` to handle a batch of queries at once: ```csharp public class GetProductBatchHandler : IBatchHandler { private readonly AppDbContext _db; public GetProductBatchHandler(AppDbContext db) => _db = db; public async ValueTask> HandleAsync( IReadOnlyList queries, CancellationToken ct) { var ids = queries.Select(q => q.ProductId).ToList(); // Single database query for all products var products = await _db.Products .Where(p => ids.Contains(p.Id)) .ToDictionaryAsync(p => p.Id, ct); return queries.ToDictionary( q => q, q => products.GetValueOrDefault(q.ProductId) ); } } ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithBatching(batching => { batching .WithMaxBatchSize(50) .WithMaxWaitTime(TimeSpan.FromMilliseconds(20)) .WithThrowIfNoBatchHandler(false) .OnBatchProcessed(info => { Console.WriteLine($"Batch of {info.BatchSize} {info.QueryType.Name} processed in {info.Duration.TotalMilliseconds}ms"); }); }); }); ``` BatchingOptions [#batchingoptions] | Option | Default | Description | | ----------------------- | ------- | ----------------------------------------------------------- | | `MaxBatchSize` | 100 | Maximum queries per batch | | `MaxWaitTime` | 10ms | Maximum time to wait for batch to fill | | `ThrowIfNoBatchHandler` | `false` | Throw if no batch handler exists (falls back to individual) | | `OnBatchProcessed` | `null` | Callback after batch processing | How It Works [#how-it-works] 1. Individual `Send()` calls with `IBatchableQuery` are queued 2. The batching behavior collects queries in a `ConcurrentQueue` 3. After `MaxWaitTime` or reaching `MaxBatchSize`, the batch is processed 4. Each query receives its individual result from the batch 5. If no `IBatchHandler` is registered, queries fall back to individual execution Practical Example [#practical-example] ```csharp // API endpoint that triggers many product lookups app.MapPost("/cart/validate", async ( CartRequest cart, IMediator mediator) => { // These queries are automatically batched into a single DB call var tasks = cart.Items.Select(async item => { var product = await mediator.Send(new GetProductByIdQuery(item.ProductId)); return new ValidatedItem(item.ProductId, product is not null, product?.Price); }); var results = await Task.WhenAll(tasks); return Results.Ok(results); }); ``` Batching is most effective when many queries of the same type are sent concurrently, such as in GraphQL resolvers or bulk validation scenarios. The behavior transparently groups queries without requiring any changes to the caller. # Caching import { Callout } from 'fumadocs-ui/components/callout'; The `TurboMediator.Caching` package provides automatic handler response caching based on attributes, avoiding redundant processing for the same input. Installation [#installation] ```bash dotnet add package TurboMediator.Caching ``` Using the Attribute [#using-the-attribute] ```csharp [Cacheable(durationSeconds: 300)] // Cache for 5 minutes public record GetProductQuery(Guid Id) : IQuery; [Cacheable(durationSeconds: 60, UseSlidingExpiration = true)] public record GetConfigQuery(string Key) : IQuery; [Cacheable(durationSeconds: 600, KeyPrefix = "catalog")] public record GetCatalogQuery(string Category) : IQuery; ``` Attribute Properties [#attribute-properties] | Property | Description | | ---------------------- | ------------------------------- | | `durationSeconds` | Cache duration | | `KeyPrefix` | Optional prefix for cache keys | | `UseSlidingExpiration` | Reset expiration on each access | Cache Key Generation [#cache-key-generation] By default, cache keys are generated using a **SHA256 hash** of the serialized message. For custom keys, implement `ICacheKeyProvider`: ```csharp public record GetUserQuery(Guid Id) : IQuery, ICacheKeyProvider { public string GetCacheKey() => $"user:{Id}"; } ``` Cache Providers [#cache-providers] In-Memory (Built-in) [#in-memory-built-in] ```csharp builder.Services.AddTurboMediator(m => { m.WithCaching, object>(); m.WithInMemoryCache(); }); ``` Redis (Official Package) [#redis-official-package] Install the Redis provider: ```bash dotnet add package TurboMediator.Caching.Redis ``` Register with a connection string: ```csharp builder.Services.AddTurboMediator(m => { m.WithCaching, object>(); m.WithRedisCache("localhost:6379"); }); ``` Or configure with full options: ```csharp builder.Services.AddTurboMediator(m => { m.WithCaching, object>(); m.WithRedisCache(options => { options.ConnectionString = "localhost:6379"; options.Database = 1; options.KeyPrefix = "myapp"; options.SerializerOptions = new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; }); }); ``` You can also reuse an existing `IConnectionMultiplexer`: ```csharp var redis = ConnectionMultiplexer.Connect("localhost:6379"); builder.Services.AddSingleton(redis); builder.Services.AddTurboMediator(m => { m.WithRedisCache(redis, options => { options.KeyPrefix = "myapp"; }); }); ``` Redis Features [#redis-features] | Feature | Description | | ------------------------ | ------------------------------------------------------------------------ | | **Key Prefix** | Isolate cache entries per application with `KeyPrefix` | | **Sliding Expiration** | Automatically renews TTL on each cache access | | **Database Selection** | Target a specific Redis database number | | **Custom Serialization** | Configure `JsonSerializerOptions` for serialization | | **Connection Reuse** | Pass an existing `IConnectionMultiplexer` to avoid duplicate connections | Custom Provider [#custom-provider] Implement `ICacheProvider`: ```csharp public interface ICacheProvider { ValueTask> GetAsync(string key, CancellationToken cancellationToken = default); ValueTask SetAsync(string key, T value, CacheEntryOptions options, CancellationToken cancellationToken = default); ValueTask RemoveAsync(string key, CancellationToken cancellationToken = default); } ``` Register it: ```csharp builder.Services.AddTurboMediator(m => { m.WithCacheProvider(); }); ``` Practical Example [#practical-example] ```csharp // Cacheable query [Cacheable(durationSeconds: 120, KeyPrefix = "products")] public record GetPopularProductsQuery(string Category, int Limit) : IQuery>, ICacheKeyProvider { public string GetCacheKey() => $"popular:{Category}:{Limit}"; } // Handler — only called on cache miss public class GetPopularProductsHandler : IQueryHandler> { private readonly IProductRepository _repo; public GetPopularProductsHandler(IProductRepository repo) => _repo = repo; public async ValueTask> Handle( GetPopularProductsQuery query, CancellationToken ct) { return await _repo.GetPopularByCategory(query.Category, query.Limit, ct); } } // Cache invalidation after write public class UpdateProductHandler : ICommandHandler { private readonly ICacheProvider _cache; public UpdateProductHandler(ICacheProvider cache) => _cache = cache; public async ValueTask Handle(UpdateProductCommand command, CancellationToken ct) { // Update product... // Invalidate cache await _cache.RemoveAsync($"products:popular:{command.Category}:{10}"); return Unit.Value; } } ``` Caching is most effective for queries (read operations). Avoid caching commands that have side effects. # Distributed Locking import { Callout } from 'fumadocs-ui/components/callout'; Distributed locking ensures that only one instance of your application processes a given operation at a time — across pods, containers, or servers. TurboMediator integrates this as a pipeline behavior, keeping your handlers free of lock management boilerplate. Distributed locking is a **concurrency** concern distinct from [Deduplication](/docs/enterprise/deduplication). Deduplication prevents the same message from being processed **twice** (idempotency). Distributed locking prevents the same **resource** from being processed **concurrently** by different instances. Packages [#packages] | Package | Purpose | | ---------------------------------------- | ------------------------------------------------------------------------------------------ | | `TurboMediator.DistributedLocking` | Core behavior, abstractions, in-memory provider | | `TurboMediator.DistributedLocking.Redis` | Redis provider via [madelson/DistributedLock](https://github.com/madelson/DistributedLock) | Installation [#installation] ```bash # Core (required) dotnet add package TurboMediator.DistributedLocking # Redis provider (production) dotnet add package TurboMediator.DistributedLocking.Redis ``` Basic Usage [#basic-usage] 1. Decorate your command [#1-decorate-your-command] ```csharp [DistributedLock] public record WithdrawCommand(Guid AccountId, decimal Amount) : ICommand, ILockKeyProvider { // Lock is scoped to this specific account — multiple accounts run concurrently. public string GetLockKey() => AccountId.ToString(); } ``` 2. Register the behavior [#2-register-the-behavior] ```csharp builder.Services.AddTurboMediator(m => m // Lock provider .WithInMemoryDistributedLocking() // development / single-node // .WithRedisDistributedLocking("localhost:6379") // production // Register the behavior for this command .WithDistributedLocking() ); ``` That's it. The handler runs only after the lock is acquired, and the lock is released automatically when the handler completes — even on exceptions. Lock Key Strategies [#lock-key-strategies] Per-entity lock (recommended) [#per-entity-lock-recommended] Implement `ILockKeyProvider` to scope the lock to a single entity, allowing unrelated entities to be processed concurrently: ```csharp [DistributedLock] public record TransferFundsCommand(Guid FromAccountId, Guid ToAccountId, decimal Amount) : ICommand, ILockKeyProvider { // Only one transfer at a time per source account public string GetLockKey() => FromAccountId.ToString(); } ``` Global message-type lock [#global-message-type-lock] Without `ILockKeyProvider`, the lock key defaults to the message type name — only one instance of this command type runs across the cluster at any given time: ```csharp [DistributedLock(TimeoutSeconds = 60)] public record RunDailyReportCommand : ICommand; // Lock key → "RunDailyReportCommand" (global for this command type) ``` Composite keys [#composite-keys] Combine multiple properties for finer granularity: ```csharp [DistributedLock(KeyPrefix = "invoice")] public record ApproveInvoiceCommand(Guid TenantId, Guid InvoiceId) : ICommand, ILockKeyProvider { public string GetLockKey() => $"{TenantId}:{InvoiceId}"; // Resulting key → "invoice:tenant-abc:invoice-123" } ``` Attribute Options [#attribute-options] ```csharp [DistributedLock( KeyPrefix = "payment", TimeoutSeconds = 15, ThrowIfNotAcquired = true )] public record ProcessPaymentCommand(Guid PaymentId) : ICommand, ILockKeyProvider { public string GetLockKey() => PaymentId.ToString(); } ``` | Property | Default | Description | | -------------------- | ----------- | -------------------------------------------------------------------------------------------------------------------- | | `KeyPrefix` | *type name* | Prefix for the lock key. Defaults to `typeof(TMessage).Name`. | | `TimeoutSeconds` | `30` | Seconds to wait for the lock before failing. Set to `0` to fail immediately. | | `ThrowIfNotAcquired` | `true` | Throw `DistributedLockException` if the lock cannot be acquired. When `false`, returns `default(TResponse)` instead. | Global Options [#global-options] Override defaults for all locking behaviors at registration time: ```csharp builder.Services.AddTurboMediator(m => m .WithInMemoryDistributedLocking() .WithDistributedLocking(options => { options.DefaultTimeout = TimeSpan.FromSeconds(10); options.GlobalKeyPrefix = "myapp"; // → "myapp:WithdrawCommand:accountId" options.DefaultThrowIfNotAcquired = true; }) ); ``` Providers [#providers] In-Memory (development / single-node) [#in-memory-development--single-node] Uses `SemaphoreSlim` per key. Zero dependencies, works without any external infrastructure. Not suitable for multi-instance deployments. ```csharp builder.Services.AddTurboMediator(m => m .WithInMemoryDistributedLocking() // ... ); ``` Redis (production) [#redis-production] Backed by [madelson/DistributedLock](https://github.com/madelson/DistributedLock), which uses the [RedLock](https://redis.io/docs/manual/patterns/distributed-locks/) algorithm under the hood. **Using a connection string (standalone setup):** ```csharp builder.Services.AddTurboMediator(m => m .WithRedisDistributedLocking("localhost:6379,abortConnect=false") // ... ); ``` **Full configuration:** ```csharp builder.Services.AddTurboMediator(m => m .WithRedisDistributedLocking(options => { options.ConnectionString = "redis-primary:6379,redis-replica:6380"; options.Database = 0; options.KeyPrefix = "turbo:locks"; }) // ... ); ``` **Reusing an existing `IConnectionMultiplexer` from DI:** ```csharp // Already registered elsewhere: builder.Services.AddSingleton( ConnectionMultiplexer.Connect("localhost:6379")); builder.Services.AddTurboMediator(m => m .WithRedisDistributedLockingFromDI(options => options.KeyPrefix = "locks") // ... ); ``` Custom Provider [#custom-provider] Implement `IDistributedLockProvider` for any backend (SQL Server, Azure Blob, etcd, …): ```csharp public class SqlDistributedLockProvider : IDistributedLockProvider { public async Task TryAcquireAsync( string key, TimeSpan timeout, CancellationToken cancellationToken) { // your implementation } } // Register: builder.Services.AddTurboMediator(m => m .WithDistributedLockProvider() // ... ); ``` Error Handling [#error-handling] When `ThrowIfNotAcquired = true` (default), a `DistributedLockException` is thrown if the lock times out: ```csharp try { var result = await mediator.Send(new WithdrawCommand(accountId, amount)); } catch (DistributedLockException ex) { // ex.LockKey → the key that could not be acquired // ex.Timeout → the timeout that was exhausted logger.LogWarning("Lock contention: {Key} waited {Timeout}s", ex.LockKey, ex.Timeout.TotalSeconds); return Conflict("This account is currently locked. Please retry."); } ``` When `ThrowIfNotAcquired = false`, `default(TResponse)` is returned and the handler is **not** called. Use this only when skipping the operation on contention is acceptable: ```csharp [DistributedLock(ThrowIfNotAcquired = false, TimeoutSeconds = 0)] public record BackgroundSyncCommand(Guid TenantId) : ICommand, ILockKeyProvider { public string GetLockKey() => TenantId.ToString(); } // Returns null instead of throwing if another sync is already running for this tenant. ``` Testing [#testing] The `InMemoryDistributedLockProvider` works perfectly in unit tests without any mocking: ```csharp builder.Services.AddTurboMediator(m => m .WithInMemoryDistributedLocking() .WithDistributedLocking() ); ``` You can also register a no-op stub if you want to isolate handler logic from locking behavior: ```csharp public class NoOpLockProvider : IDistributedLockProvider { private sealed class AlwaysAcquiredHandle : IDistributedLockHandle { public string Key { get; init; } = string.Empty; public ValueTask DisposeAsync() => ValueTask.CompletedTask; } public Task TryAcquireAsync( string key, TimeSpan timeout, CancellationToken ct) => Task.FromResult(new AlwaysAcquiredHandle { Key = key }); } ``` When to Use Distributed Locking [#when-to-use-distributed-locking] | Scenario | Why it helps | | ----------------------------------------------- | ---------------------------------------------------------------- | | Account balance operations (withdraw, transfer) | Prevents double-spending from concurrent requests | | Inventory reservation | Prevents over-selling the same stock unit | | Scheduled job execution | Ensures only one instance runs a job at a time | | Saga / saga step execution | Prevents two instances from advancing the same saga concurrently | | External API calls with side effects | Avoids duplicated charges or state changes | Distributed locking introduces latency and a dependency on an external coordination service. Use it only for **write operations** on shared mutable state. Read-only queries should never need a lock. Sample [#sample] See [`samples/Sample.DistributedLocking`](https://github.com/marcocestari/TurboMediator/tree/main/samples/Sample.DistributedLocking) for a complete bank account API demonstrating per-account locking, concurrent stress-testing, and lock contention handling. # Feature Flags import { Callout } from 'fumadocs-ui/components/callout'; The feature flag behavior allows you to conditionally enable or disable message handlers based on feature flags, without deploying new code. Installation [#installation] ```bash dotnet add package TurboMediator.FeatureFlags # Optional: Microsoft.FeatureManagement integration dotnet add package TurboMediator.FeatureFlags.FeatureManagement ``` Using the Attribute [#using-the-attribute] ```csharp [FeatureFlag("new-checkout")] public record NewCheckoutCommand(Guid CartId) : ICommand; [FeatureFlag("beta-search", FallbackBehavior = FeatureFallback.ReturnDefault)] public record BetaSearchQuery(string Term) : IQuery; [FeatureFlag("premium-export", FallbackBehavior = FeatureFallback.Throw)] public record ExportReportCommand(Guid ReportId) : ICommand; [FeatureFlag("per-user-feature", PerUser = true)] public record PersonalizedFeedQuery : IQuery; ``` Attribute Properties [#attribute-properties] | Property | Default | Description | | ------------------ | -------- | ------------------------------------------------------------ | | Feature name | Required | The feature flag name | | `FallbackBehavior` | `Throw` | What to do when flag is disabled: Throw, ReturnDefault, Skip | | `PerUser` | `false` | Evaluate per-user instead of globally | IFeatureFlagProvider [#ifeatureflagprovider] ```csharp public interface IFeatureFlagProvider { ValueTask IsEnabledAsync(string featureName, CancellationToken cancellationToken = default); ValueTask IsEnabledAsync(string featureName, string userId, CancellationToken cancellationToken = default); } ``` In-Memory Provider [#in-memory-provider] For development and testing: ```csharp builder.Services.AddTurboMediator(m => { m.WithFeatureFlags(ff => { ff.UseInMemoryProvider() .WithFeature("new-checkout", true) .WithFeature("beta-search", false) .WithFeature("premium-export", enabled: true); }); }); ``` Microsoft.FeatureManagement Integration [#microsoftfeaturemanagement-integration] Integrate with `Microsoft.FeatureManagement` for configuration-driven flags: ```bash dotnet add package TurboMediator.FeatureFlags.FeatureManagement ``` ```csharp // appsettings.json { "FeatureManagement": { "new-checkout": true, "beta-search": { "EnabledFor": [ { "Name": "Percentage", "Parameters": { "Value": 50 } } ] } } } // Registration builder.Services.AddFeatureManagement(); builder.Services.AddMicrosoftFeatureFlags(); builder.Services.AddTurboMediator(m => { m.WithFeatureFlags(ff => { // FeatureManagement provider is auto-detected }); }); ``` FeatureFlagOptions [#featureflagoptions] ```csharp builder.Services.AddTurboMediator(m => { m.WithFeatureFlags(ff => { ff.UseInMemoryProvider() .UseStrictMode() // or .UseLenientMode() .WithUserIdProvider(() => currentUserService.GetUserId()); }); }); ``` | Option | Description | | ----------------------------------- | -------------------------------------------------------- | | `UseStrictMode()` | Sets default fallback to `FeatureFallback.Throw` | | `UseLenientMode()` | Sets default fallback to `FeatureFallback.ReturnDefault` | | `WithUserIdProvider(Func)` | Parameterless function to get user ID | Custom Provider [#custom-provider] ```csharp public class LaunchDarklyProvider : IFeatureFlagProvider { private readonly ILdClient _client; public LaunchDarklyProvider(ILdClient client) => _client = client; public ValueTask IsEnabledAsync(string featureName) { return new ValueTask( _client.BoolVariation(featureName, Context.Default, false)); } public ValueTask IsEnabledAsync(string featureName, string userId) { var context = Context.Builder(userId).Build(); return new ValueTask( _client.BoolVariation(featureName, context, false)); } } ``` Feature flags are evaluated at the pipeline level, so disabled features never reach the handler. This provides a clean way to do gradual rollouts, A/B testing, and kill switches. # FluentValidation import { Callout } from 'fumadocs-ui/components/callout'; The FluentValidation integration automatically validates messages before they reach the handler, using your existing FluentValidation validators. Installation [#installation] ```bash dotnet add package TurboMediator.FluentValidation ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { // Auto-discover validators in the current assembly m.WithFluentValidation(); // Or specify the assembly m.WithFluentValidation(); }); ``` Creating Validators [#creating-validators] Define validators using standard FluentValidation: ```csharp using FluentValidation; public class CreateUserCommandValidator : AbstractValidator { public CreateUserCommandValidator() { RuleFor(x => x.Name) .NotEmpty().WithMessage("Name is required") .MaximumLength(100).WithMessage("Name must be 100 characters or less"); RuleFor(x => x.Email) .NotEmpty().WithMessage("Email is required") .EmailAddress().WithMessage("Invalid email format"); RuleFor(x => x.Age) .InclusiveBetween(18, 120).WithMessage("Age must be between 18 and 120"); } } public class UpdateProductCommandValidator : AbstractValidator { private readonly IProductRepository _repository; public UpdateProductCommandValidator(IProductRepository repository) { _repository = repository; RuleFor(x => x.Name) .NotEmpty() .MustAsync(async (name, ct) => { var existing = await _repository.FindByNameAsync(name, ct); return existing is null; }).WithMessage("Product name already exists"); RuleFor(x => x.Price) .GreaterThan(0).WithMessage("Price must be positive"); } } ``` ValidationException [#validationexception] When validation fails, a `ValidationException` is thrown with all failures: ```csharp public class ValidationException : Exception { public IReadOnlyList Failures { get; } } public class ValidationFailure { public string PropertyName { get; } public string ErrorMessage { get; } public object? AttemptedValue { get; } public string? ErrorCode { get; } public ValidationSeverity Severity { get; } } public enum ValidationSeverity { Error, Warning, Info } ``` Handling Validation Errors [#handling-validation-errors] ```csharp app.MapPost("/users", async (CreateUserCommand command, IMediator mediator) => { try { var user = await mediator.Send(command); return Results.Created($"/users/{user.Id}", user); } catch (ValidationException ex) { var errors = ex.Failures .Select(f => new { f.PropertyName, f.ErrorMessage }) .ToList(); return Results.BadRequest(new { Errors = errors }); } }); ``` Global Exception Handler [#global-exception-handler] ```csharp app.UseExceptionHandler(app => { app.Run(async context => { var exception = context.Features.Get()?.Error; if (exception is ValidationException validationEx) { context.Response.StatusCode = 400; await context.Response.WriteAsJsonAsync(new { Type = "ValidationError", Errors = validationEx.Failures.Select(f => new { Field = f.PropertyName, Message = f.ErrorMessage, Severity = f.Severity.ToString() }) }); } }); }); ``` Multiple Validators [#multiple-validators] You can define multiple validators for the same message. All validators are executed and all failures are collected: ```csharp // Basic validation rules public class CreateOrderBasicValidator : AbstractValidator { public CreateOrderBasicValidator() { RuleFor(x => x.Items).NotEmpty(); } } // Business rule validation (separate concerns) public class CreateOrderBusinessValidator : AbstractValidator { public CreateOrderBusinessValidator(IInventoryService inventory) { RuleForEach(x => x.Items).MustAsync(async (item, ct) => { var stock = await inventory.GetStockAsync(item.ProductId, ct); return stock >= item.Quantity; }).WithMessage("Insufficient stock"); } } ``` The `FluentValidationBehavior` resolves **all** `IValidator` from DI and runs them. Failures from all validators are aggregated into a single `ValidationException`. # Rate Limiting import { Callout } from 'fumadocs-ui/components/callout'; The rate limiting package provides two complementary behaviors: **Rate Limiting** to control request throughput, and **Bulkhead Isolation** to limit concurrent executions. Installation [#installation] ```bash dotnet add package TurboMediator.RateLimiting ``` Rate Limiting [#rate-limiting] Using the Attribute [#using-the-attribute] ```csharp [RateLimit(maxRequests: 10, windowSeconds: 60)] public record SendSmsCommand(string Phone, string Message) : ICommand; [RateLimit(maxRequests: 100, windowSeconds: 60, PerUser = true)] public record SearchQuery(string Term) : IQuery; [RateLimit(maxRequests: 50, windowSeconds: 60, PerTenant = true)] public record ApiCallCommand(string Endpoint) : ICommand; ``` Using Configuration [#using-configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithRateLimiting(options => { options.MaxRequests = 10; options.WindowSeconds = 60; options.Algorithm = RateLimiterAlgorithm.SlidingWindow; options.PerUser = true; options.ThrowOnRateLimitExceeded = true; }); // Or apply globally m.WithGlobalRateLimiting(options => { options.MaxRequests = 1000; options.WindowSeconds = 60; }); }); ``` RateLimitOptions [#ratelimitoptions] | Option | Default | Description | | -------------------------- | ---------------------------------- | --------------------------------------------------------------- | | `MaxRequests` | 100 | Max requests per window | | `WindowSeconds` | 60 | Time window in seconds | | `PerUser` | `false` | Per-user rate limiting | | `PerTenant` | `false` | Per-tenant rate limiting | | `PerIpAddress` | `false` | Per-IP rate limiting | | `Algorithm` | `RateLimiterAlgorithm.FixedWindow` | Algorithm: FixedWindow, SlidingWindow, TokenBucket, Concurrency | | `TokensPerPeriod` | 10 | Tokens per period (TokenBucket algorithm) | | `SegmentsPerWindow` | 4 | Segments per window (SlidingWindow algorithm) | | `QueueProcessingOrder` | `OldestFirst` | Order for processing queued requests | | `PolicyName` | `null` | Named policy identifier | | `QueueExceededRequests` | `false` | Queue instead of reject | | `MaxQueueSize` | 0 | Maximum queue size | | `ThrowOnRateLimitExceeded` | `true` | Throw vs return default | | `UserIdProvider` | `null` | Custom user ID provider | | `TenantIdProvider` | `null` | Custom tenant ID provider | | `IpAddressProvider` | `null` | Custom IP provider | | `CustomRateLimiterFactory` | `null` | Custom rate limiter | Algorithms [#algorithms] | Algorithm | Description | | --------------- | ----------------------------------------- | | `FixedWindow` | Fixed time windows (e.g., 100 req/min) | | `SlidingWindow` | Rolling time window for smoother limiting | | `TokenBucket` | Token-based with steady refill rate | | `Concurrency` | Limits concurrent executions | Convenience Methods [#convenience-methods] ```csharp m.WithSlidingWindowRateLimit( maxRequests: 50, windowSeconds: 60, segmentsPerWindow: 4); m.WithTokenBucketRateLimit( bucketSize: 100, tokensPerPeriod: 10, replenishmentPeriodSeconds: 60); m.WithPerUserRateLimit( maxRequestsPerUser: 10, windowSeconds: 60, userIdProvider: () => GetCurrentUserId()); ``` RateLimitExceededException [#ratelimitexceededexception] ```csharp try { await mediator.Send(new SendSmsCommand("555-1234", "Hello")); } catch (RateLimitExceededException ex) { // ex.MessageType — what was rate limited // ex.PartitionKey — the partition that exceeded limits // ex.RetryAfter — suggested retry delay return Results.StatusCode(429); } ``` *** Bulkhead Isolation [#bulkhead-isolation] Limits **concurrent execution** of handlers to prevent resource exhaustion. Using the Attribute [#using-the-attribute-1] ```csharp [Bulkhead(maxConcurrent: 5, maxQueue: 10)] public record ProcessVideoCommand(string VideoUrl) : ICommand; ``` Using Configuration [#using-configuration-1] ```csharp builder.Services.AddTurboMediator(m => { m.WithBulkhead(options => { options.MaxConcurrent = 5; options.MaxQueue = 10; options.QueueTimeout = TimeSpan.FromSeconds(30); options.ThrowOnBulkheadFull = true; options.TrackMetrics = true; options.OnRejection = (info) => { Console.WriteLine($"Bulkhead full for {info.MessageType}: {info.Reason}"); }; }); // Or global bulkhead m.WithGlobalBulkhead(options => { options.MaxConcurrent = 50; }); }); ``` BulkheadOptions [#bulkheadoptions] | Option | Default | Description | | ---------------------- | ------- | -------------------------------- | | `MaxConcurrent` | 10 | Maximum concurrent executions | | `MaxQueue` | 100 | Maximum queued requests | | `QueueTimeout` | `null` | Timeout waiting in queue | | `ThrowOnBulkheadFull` | `true` | Throw `BulkheadFullException` | | `TrackMetrics` | `false` | Track concurrency metrics | | `PerPartition` | `false` | Separate bulkheads per partition | | `PartitionKeyProvider` | `null` | Custom partition key | | `OnRejection` | `null` | Callback when rejected | Combined Rate Limiting + Bulkhead [#combined-rate-limiting--bulkhead] ```csharp builder.Services.AddTurboMediator(m => { // Apply both at once m.WithThrottling( rateLimitOptions => { rateLimitOptions.MaxRequests = 100; rateLimitOptions.WindowSeconds = 60; }, bulkheadOptions => { bulkheadOptions.MaxConcurrent = 5; }); }); ``` Rate limiting controls **throughput** (requests per time window), while bulkhead isolation controls **concurrency** (simultaneous executions). Use both together for comprehensive resource protection. # Result Pattern import { Callout } from 'fumadocs-ui/components/callout'; The `TurboMediator.Result` package provides `Result` and `Result` types for explicit error handling without relying on exceptions for control flow. Installation [#installation] ```bash dotnet add package TurboMediator.Result ``` ```csharp using TurboMediator.Results; ``` Result [#resultt] A readonly struct that represents either a successful value or an error: ```csharp public readonly struct Result { public bool IsSuccess { get; } public bool IsFailure { get; } public TValue Value { get; } // throws if failure public Exception Error { get; } // throws if success } ``` Creating Results [#creating-results] ```csharp // Success var success = Result.Success(42); var success2 = Result.Success(new User("Alice")); // Failure var failure = Result.Failure(new InvalidOperationException("oops")); // From try-catch var result = Result.Try(() => int.Parse("abc")); // Failure with FormatException var asyncResult = await Result.TryAsync(() => httpClient.GetStringAsync(url)); ``` Implicit Conversions [#implicit-conversions] ```csharp // Value → Success result Result result = 42; // Exception → Failure result Result error = new InvalidOperationException("error"); ``` Pattern Matching [#pattern-matching] ```csharp var result = await mediator.Send(new GetUserQuery(userId)); // Match var message = result.Match( onSuccess: user => $"Found {user.Name}", onFailure: error => $"Error: {error.Message}" ); // Map — transform the success value var nameResult = result.Map(user => user.Name); // Bind — chain operations var orderResult = result.Bind(user => GetOrdersForUser(user.Id)); ``` Accessing Values [#accessing-values] ```csharp // Safe access with default var user = result.GetValueOrDefault(User.Guest); // Throw if failure var user = result.GetValueOrThrow(); // throws the stored exception if failure ``` Result [#resulttvalue-terror] A typed error variant for domain-specific errors without using exceptions: ```csharp public readonly struct Result { public bool IsSuccess { get; } public bool IsFailure { get; } public TValue Value { get; } public TError Error { get; } } ``` Practical Example [#practical-example] ```csharp // Define domain errors public abstract record UserError { public record NotFound(Guid Id) : UserError; public record EmailTaken(string Email) : UserError; public record ValidationFailed(string[] Errors) : UserError; } // Use Result in handler public record CreateUserCommand(string Name, string Email) : ICommand>; public class CreateUserHandler : ICommandHandler> { private readonly IUserRepository _repo; public CreateUserHandler(IUserRepository repo) => _repo = repo; public async ValueTask> Handle( CreateUserCommand command, CancellationToken ct) { if (string.IsNullOrWhiteSpace(command.Name)) return new UserError.ValidationFailed(new[] { "Name is required" }); var existingUser = await _repo.FindByEmailAsync(command.Email, ct); if (existingUser is not null) return new UserError.EmailTaken(command.Email); var user = new User(Guid.NewGuid(), command.Name, command.Email); await _repo.AddAsync(user, ct); return user; // implicit conversion to success } } // Usage in API endpoint app.MapPost("/users", async (CreateUserCommand cmd, IMediator mediator) => { var result = await mediator.Send(cmd); return result.Match( onSuccess: user => Results.Created($"/users/{user.Id}", user), onFailure: error => error switch { UserError.EmailTaken e => Results.Conflict($"Email {e.Email} is taken"), UserError.ValidationFailed v => Results.BadRequest(v.Errors), _ => Results.StatusCode(500) } ); }); ``` The Result pattern is especially useful in domain-driven design where you want to represent expected failures (validation errors, not found, business rule violations) without throwing exceptions. # Saga Pattern import { Callout } from 'fumadocs-ui/components/callout'; The saga pattern manages distributed transactions by breaking them into a sequence of steps, each with a compensating action that undoes the step if a later step fails. Installation [#installation] ```bash dotnet add package TurboMediator.Saga dotnet add package TurboMediator.Saga.EntityFramework # Optional: EF Core store ``` Defining a Saga [#defining-a-saga] Extend `Saga` and define steps in the constructor: ```csharp public class OrderSagaData { public Guid OrderId { get; set; } public Guid PaymentId { get; set; } public Guid ShipmentId { get; set; } public string CustomerEmail { get; set; } = ""; } public class OrderSaga : Saga { public OrderSaga() { AddStep(Step("CreateOrder") .Execute( data => new CreateOrderCommand(data.CustomerEmail), (data, result) => data.OrderId = result.Id) .Compensate( data => new CancelOrderCommand(data.OrderId))); AddStep(Step("ProcessPayment") .Execute( data => new ProcessPaymentCommand(data.OrderId, 99.99m), (data, result) => data.PaymentId = result.PaymentId) .Compensate( data => new RefundPaymentCommand(data.PaymentId))); AddStep(Step("ArrangeShipment") .Execute( data => new ArrangeShipmentCommand(data.OrderId), (data, result) => data.ShipmentId = result.ShipmentId) .Compensate( data => new CancelShipmentCommand(data.ShipmentId))); AddStep(Step("SendConfirmation") .Execute( data => new SendOrderConfirmationCommand(data.CustomerEmail, data.OrderId))); // No compensation for email — can't unsend it } } ``` SagaStepBuilder API [#sagastepbuilder-api] | Method | Description | | ---------------------------------------------------------------------- | ------------------------------------------------------- | | `Step(name)` | Creates a `SagaStepBuilder` for a named step | | `AddStep(builder)` | Registers the step builder into the saga | | `.Execute(factory, onSuccess?)` | Define the step's forward action via a mediator command | | `.Execute(Func>)` | Custom async execution function | | `.Compensate(factory)` | Define the compensating action via a mediator command | | `.Compensate(Func)` | Custom async compensation function | Executing Sagas [#executing-sagas] Use `SagaOrchestrator`: ```csharp var orchestrator = serviceProvider.GetRequiredService>(); var saga = new OrderSaga(); var result = await orchestrator.ExecuteAsync( saga, new OrderSagaData { CustomerEmail = "john@example.com" }, correlationId: Guid.NewGuid().ToString()); ``` SagaResult [#sagaresult] ```csharp public class SagaResult { public Guid SagaId { get; } public bool IsSuccess { get; } public TData? Data { get; } public string? Error { get; } public IReadOnlyList CompensationErrors { get; } } ``` Saga Status [#saga-status] ```csharp public enum SagaStatus { NotStarted, // Initial state Running, // Executing steps Completed, // All steps completed Failed, // A step failed Compensating, // Running compensating actions Compensated // All compensations completed } ``` Compensation Flow [#compensation-flow] When a step fails, the orchestrator runs compensating actions **in reverse order**: ``` Step 1 ✓ → Step 2 ✓ → Step 3 ✗ │ Compensate 2 ← Compensate 3 │ Compensate 1 ``` Resuming Interrupted Sagas [#resuming-interrupted-sagas] If the application crashes during saga execution, you can resume: ```csharp var result = await orchestrator.ResumeAsync(saga, sagaId); ``` Saga Stores [#saga-stores] In-Memory (built-in) [#in-memory-built-in] ```csharp builder.Services.AddTurboMediator(m => { m.WithSagas(saga => { saga.UseInMemoryStore(); saga.AddOrchestrator(); }); }); // Or shorthand builder.Services.AddTurboMediator(m => m.WithInMemorySagas()); ``` Entity Framework Core [#entity-framework-core] ```bash dotnet add package TurboMediator.Saga.EntityFramework ``` ```csharp // DbContext setup protected override void OnModelCreating(ModelBuilder modelBuilder) { modelBuilder.ApplySagaStateConfiguration(); } // Registration builder.Services.AddTurboMediator(m => { m.WithSagas(saga => { saga.UseStore>(); saga.AddOrchestrator(); }); // Or register directly on services // builder.Services.AddEfCoreSagaStore(); }); ``` Complete Example [#complete-example] ```csharp // 1. Define saga data public class BookingData { public Guid FlightId { get; set; } public Guid HotelId { get; set; } public Guid CarId { get; set; } } // 2. Define the saga public class TravelBookingSaga : Saga { public TravelBookingSaga() { AddStep(Step("BookFlight") .Execute( d => new BookFlightCommand(d.FlightId), (d, r) => d.FlightId = r.BookingId) .Compensate( d => new CancelFlightCommand(d.FlightId))); AddStep(Step("BookHotel") .Execute( d => new BookHotelCommand(d.HotelId), (d, r) => d.HotelId = r.BookingId) .Compensate( d => new CancelHotelCommand(d.HotelId))); AddStep(Step("BookCar") .Execute( d => new BookCarCommand(d.CarId), (d, r) => d.CarId = r.BookingId) .Compensate( d => new CancelCarCommand(d.CarId))); } } // 3. Execute var result = await orchestrator.ExecuteAsync( new TravelBookingSaga(), new BookingData { FlightId = flightId, HotelId = hotelId, CarId = carId }, correlationId: Guid.NewGuid().ToString()); if (result.IsSuccess) Console.WriteLine($"Booking completed: {result.SagaId}"); else Console.WriteLine($"Booking failed: {result.Error}"); ``` Sagas provide **eventual consistency**, not ACID transactions. Each step commits independently. Design compensating actions carefully to handle partial states. # State Machine import { Callout } from 'fumadocs-ui/components/callout'; The state machine package models **entities with well-defined lifecycles** (Order, Payment, Ticket, Subscription, etc.) where state transitions are explicit, validated, and auditable. **State Machine vs Saga**: Sagas orchestrate multi-step distributed processes. State machines govern the lifecycle of a single entity over time with strict transition rules. Installation [#installation] ```bash dotnet add package TurboMediator.StateMachine dotnet add package TurboMediator.StateMachine.EntityFramework # Optional: EF Core transition store ``` Concepts [#concepts] | Concept | Description | | --------------------- | ------------------------------------------------------------ | | **State** | A named position in the entity lifecycle (enum value) | | **Trigger** | An event that causes a transition (enum value) | | **Guard** | A condition that must be true for a transition to be allowed | | **Entry/Exit Action** | Code that runs when entering or leaving a state | | **Final State** | A state with no outgoing transitions (terminal) | | **Transition Store** | Persists transition history for auditing | Defining States and Triggers [#defining-states-and-triggers] Use enums to define all possible states and triggers: ```csharp public enum OrderStatus { Draft, Submitted, Approved, Shipped, Delivered, Cancelled } public enum OrderTrigger { Submit, Approve, Reject, Ship, Deliver, Cancel } ``` Making an Entity Stateful [#making-an-entity-stateful] Implement `IStateful` on your entity: ```csharp public class Order : IStateful { public Guid Id { get; set; } = Guid.NewGuid(); public OrderStatus CurrentState { get; set; } = OrderStatus.Draft; public string CustomerName { get; set; } = ""; public decimal Total { get; set; } public DateTime? ApprovedAt { get; set; } public string? CancellationReason { get; set; } } ``` Defining a State Machine [#defining-a-state-machine] Extend `StateMachine` and override `Configure`: ```csharp public class OrderStateMachine : StateMachine { public OrderStateMachine(IMediator mediator, ITransitionStore? transitionStore = null) : base(mediator, transitionStore) { } protected override void Configure(StateMachineBuilder builder) { builder.InitialState(OrderStatus.Draft); builder.State(OrderStatus.Draft) .Permit(OrderTrigger.Submit, OrderStatus.Submitted) .When(order => order.Total > 0, "Total > 0") .When(order => order.Items.Count > 0, "Has items") .Permit(OrderTrigger.Cancel, OrderStatus.Cancelled); builder.State(OrderStatus.Submitted) .OnEntry(async (order, ctx) => { await ctx.Publish(new OrderSubmittedEvent(order.Id)); }) .Permit(OrderTrigger.Approve, OrderStatus.Approved) .When(order => order.Total < 50_000, "Auto-approve limit") .Permit(OrderTrigger.Reject, OrderStatus.Draft) .Permit(OrderTrigger.Cancel, OrderStatus.Cancelled); builder.State(OrderStatus.Approved) .OnEntry(async (order, ctx) => { order.ApprovedAt = DateTime.UtcNow; await ctx.Send(new ReserveInventoryCommand(order.Id)); }) .Permit(OrderTrigger.Ship, OrderStatus.Shipped); builder.State(OrderStatus.Shipped) .Permit(OrderTrigger.Deliver, OrderStatus.Delivered); builder.State(OrderStatus.Delivered) .AsFinal(); builder.State(OrderStatus.Cancelled) .OnEntry(async (order, ctx) => { if (ctx.Metadata.TryGetValue("reason", out var reason)) order.CancellationReason = reason; await ctx.Publish(new OrderCancelledEvent(order.Id, reason)); }) .AsFinal(); builder.OnTransition(async (order, from, to, trigger) => { // Global audit callback for every transition logger.LogInformation("Order {Id}: {From} → {To} via {Trigger}", order.Id, from, to, trigger); await Task.CompletedTask; }); } } ``` Fluent API Reference [#fluent-api-reference] StateMachineBuilder [#statemachinebuilder] | Method | Description | | ------------------------------- | ------------------------------------------------------- | | `.InitialState(state)` | Sets the initial state for new entities | | `.State(state)` | Begins configuration for a specific state | | `.OnTransition(callback)` | Global callback invoked on every successful transition | | `.OnInvalidTransition(handler)` | Custom handler for disallowed triggers (default: throw) | StateConfiguration [#stateconfiguration] | Method | Description | | ------------------------------- | ------------------------------------------------- | | `.Permit(trigger, destination)` | Allows a transition via the specified trigger | | `.OnEntry(action)` | Action to run when entering this state | | `.OnExit(action)` | Action to run when leaving this state | | `.AsFinal()` | Marks as terminal state (no outgoing transitions) | TransitionBuilder [#transitionbuilder] | Method | Description | | ------------------------------- | ------------------------------------------------------------- | | `.When(guard, description?)` | Adds a guard condition (chainable for multiple guards) | | `.Permit(trigger, destination)` | Shortcut to add another transition from the same source state | Registration [#registration] Fluent Builder (recommended) [#fluent-builder-recommended] ```csharp builder.Services.AddTurboMediator(m => { m.WithInMemoryStateMachines(sm => { sm.AddStateMachine(); sm.AddStateMachine(); }); }); ``` Manual Registration [#manual-registration] ```csharp builder.Services.AddTurboMediator(m => { m.WithStateMachines(sm => { sm.UseStore(); // Custom store sm.AddStateMachine(); }); }); ``` Direct IServiceCollection [#direct-iservicecollection] ```csharp builder.Services.AddInMemoryTransitionStore(); builder.Services.AddStateMachine(); ``` Firing Transitions [#firing-transitions] Inject `IStateMachine` and use `FireAsync`: ```csharp public class OrderController(IStateMachine machine) { [HttpPost("{id}/submit")] public async Task Submit(Guid id) { var order = await GetOrder(id); var result = await machine.FireAsync(order, OrderTrigger.Submit); if (!result.IsSuccess) return BadRequest(result.Error); await SaveOrder(order); // Persist the updated state return Ok(new { result.PreviousState, result.CurrentState, result.Trigger, result.Timestamp }); } [HttpPost("{id}/cancel")] public async Task Cancel(Guid id, [FromBody] CancelRequest request) { var order = await GetOrder(id); // Pass metadata to entry actions var metadata = new Dictionary { ["reason"] = request.Reason }; var result = await machine.FireAsync(order, OrderTrigger.Cancel, metadata); if (!result.IsSuccess) return BadRequest(result.Error); await SaveOrder(order); return Ok(result); } } ``` TransitionResult [#transitionresult] Every `FireAsync` call returns a `TransitionResult`: ```csharp public class TransitionResult { public TState PreviousState { get; } public TState CurrentState { get; } public TTrigger Trigger { get; } public bool IsSuccess { get; } public string? Error { get; } // Null on success; guard/transition error on failure public DateTime Timestamp { get; } } ``` When a guard fails, `FireAsync` returns a failed `TransitionResult` instead of throwing. When a trigger is not defined for the current state, an `InvalidTransitionException` is thrown (unless you configure `OnInvalidTransition`). TransitionContext [#transitioncontext] Entry and exit actions receive a `TransitionContext` that provides access to the mediator: ```csharp .OnEntry(async (entity, ctx) => { // Send commands await ctx.Send(new ReserveInventoryCommand(entity.Id)); // Publish notifications await ctx.Publish(new OrderApprovedEvent(entity.Id)); // Access metadata passed by the caller var reason = ctx.Metadata["reason"]; }) ``` Guard Conditions [#guard-conditions] Guards prevent transitions unless conditions are met. They are evaluated **before** entry/exit actions: ```csharp builder.State(OrderStatus.Draft) .Permit(OrderTrigger.Submit, OrderStatus.Submitted) .When(order => order.Total > 0, "Total must be positive") .When(order => order.Items.Count > 0, "Must have items"); ``` Multiple `.When()` calls on the same transition are AND-combined — all must pass. Querying State [#querying-state] ```csharp // Get all triggers valid in the current state (evaluates guards) IReadOnlyList triggers = machine.GetPermittedTriggers(order); // Check a specific trigger bool canSubmit = machine.CanFire(order, OrderTrigger.Submit); // Check if a state is terminal bool isFinal = machine.IsFinalState(OrderStatus.Delivered); // Get all configured states IReadOnlyList states = machine.GetAllStates(); // Get all transitions (for rendering) var transitions = machine.GetAllTransitions(); // Returns: List<(Source, Trigger, Destination, GuardDescriptions)> ``` Final States [#final-states] Final states cannot have outgoing transitions. Attempting to fire a trigger on an entity in a final state throws `InvalidTransitionException`: ```csharp builder.State(OrderStatus.Delivered) .AsFinal(); // No .Permit() calls allowed after this builder.State(OrderStatus.Cancelled) .OnEntry(async (entity, ctx) => { /* cleanup */ }) .AsFinal(); ``` Invalid Transition Handling [#invalid-transition-handling] By default, firing an undefined trigger throws `InvalidTransitionException`: ```csharp try { await machine.FireAsync(order, OrderTrigger.Ship); // Not in Draft } catch (InvalidTransitionException ex) { // ex.CurrentState = "Draft" // ex.Trigger = "Ship" // ex.EntityType = "Order" } ``` You can override this with a custom handler: ```csharp builder.OnInvalidTransition((entity, trigger) => { logger.LogWarning("Invalid trigger {Trigger} on {State}", trigger, entity.CurrentState); // Does not throw — returns a failed TransitionResult instead }); ``` Mermaid Diagrams [#mermaid-diagrams] State machines can generate Mermaid state diagrams for documentation: ```csharp if (machine is OrderStateMachine osm) { string diagram = osm.ToMermaidDiagram(); Console.WriteLine(diagram); } ``` Output: ```mermaid stateDiagram-v2 [*] --> Draft Draft --> Submitted : Submit [Total > 0, Has items] Draft --> Cancelled : Cancel Submitted --> Approved : Approve [Auto-approve limit] Submitted --> Draft : Reject Submitted --> Cancelled : Cancel Approved --> Shipped : Ship Shipped --> Delivered : Deliver Delivered --> [*] Cancelled --> [*] ``` Transition Stores [#transition-stores] Transition stores persist an audit trail of all state changes for compliance, debugging, and analytics. In-Memory (built-in) [#in-memory-built-in] ```csharp m.WithInMemoryStateMachines(sm => { sm.AddStateMachine(); }); ``` Entity Framework Core [#entity-framework-core] ```bash dotnet add package TurboMediator.StateMachine.EntityFramework ``` ```csharp // DbContext setup protected override void OnModelCreating(ModelBuilder modelBuilder) { modelBuilder.ApplyTransitionConfiguration(); // Or with options: modelBuilder.ApplyTransitionConfiguration(new EfCoreTransitionStoreOptions { TableName = "StateTransitions", SchemaName = "audit", UseJsonColumn = true }); } // Registration builder.Services.AddEfCoreTransitionStore(options => { options.TableName = "StateTransitions"; options.AutoMigrate = true; }); ``` EfCoreTransitionStoreOptions [#efcoretransitionstoreoptions] | Property | Default | Description | | --------------- | -------------------- | --------------------------------- | | `TableName` | `"StateTransitions"` | Table name for transition records | | `SchemaName` | `null` | Schema (null = default schema) | | `AutoMigrate` | `false` | Auto-create table on first use | | `UseJsonColumn` | `false` | Use JSON column type for metadata | Querying History [#querying-history] ```csharp var store = serviceProvider.GetRequiredService(); await foreach (var record in store.GetHistoryAsync("order-123")) { Console.WriteLine($"{record.Timestamp}: {record.FromState} → {record.ToState} via {record.Trigger}"); } ``` Complete Example [#complete-example] ```csharp // --- States & Triggers --- public enum TicketStatus { Open, InProgress, Review, Done, Closed } public enum TicketTrigger { Assign, StartWork, RequestReview, Approve, Reject, Close } // --- Entity --- public class Ticket : IStateful { public Guid Id { get; set; } = Guid.NewGuid(); public TicketStatus CurrentState { get; set; } = TicketStatus.Open; public string Title { get; set; } = ""; public string? AssignedTo { get; set; } } // --- State Machine --- public class TicketStateMachine : StateMachine { public TicketStateMachine(IMediator mediator, ITransitionStore? store = null) : base(mediator, store) { } protected override void Configure(StateMachineBuilder builder) { builder.InitialState(TicketStatus.Open); builder.State(TicketStatus.Open) .Permit(TicketTrigger.Assign, TicketStatus.InProgress) .When(t => !string.IsNullOrEmpty(t.AssignedTo), "Must be assigned") .Permit(TicketTrigger.Close, TicketStatus.Closed); builder.State(TicketStatus.InProgress) .Permit(TicketTrigger.RequestReview, TicketStatus.Review); builder.State(TicketStatus.Review) .Permit(TicketTrigger.Approve, TicketStatus.Done) .Permit(TicketTrigger.Reject, TicketStatus.InProgress); builder.State(TicketStatus.Done) .Permit(TicketTrigger.Close, TicketStatus.Closed); builder.State(TicketStatus.Closed) .AsFinal(); } } // --- Registration --- builder.Services.AddTurboMediator(m => { m.WithInMemoryStateMachines(sm => { sm.AddStateMachine(); }); }); // --- Usage --- var machine = serviceProvider.GetRequiredService>(); var ticket = new Ticket { Title = "Fix login bug", AssignedTo = "alice" }; await machine.FireAsync(ticket, TicketTrigger.Assign); // Open → InProgress await machine.FireAsync(ticket, TicketTrigger.RequestReview); // InProgress → Review await machine.FireAsync(ticket, TicketTrigger.Approve); // Review → Done await machine.FireAsync(ticket, TicketTrigger.Close); // Done → Closed Console.WriteLine(ticket.CurrentState); // Closed ``` The state machine mutates the entity's `CurrentState` property in-memory. You are responsible for persisting the entity to your database after a successful transition. # Built-in Validation The `TurboMediator.Validation` package provides a lightweight built-in validation system with an `AbstractValidator` and `RuleBuilder`, independent of FluentValidation. Installation [#installation] ```bash dotnet add package TurboMediator.Validation ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithValidation(); }); ``` Creating Validators [#creating-validators] ```csharp public class CreateUserValidator : AbstractValidator { public CreateUserValidator() { RuleFor(x => x.Name) .NotEmpty() .MaximumLength(100); RuleFor(x => x.Email) .NotEmpty() .EmailAddress(); } } ``` Register validators in DI: ```csharp builder.Services.AddTransient, CreateUserValidator>(); ``` Difference from FluentValidation [#difference-from-fluentvalidation] This built-in validator is a **lightweight alternative** for projects that don't want the FluentValidation dependency. For full-featured validation (complex rules, conditional validation, custom validators), use the `TurboMediator.FluentValidation` package instead. # Audit import { Callout } from 'fumadocs-ui/components/callout'; The audit behavior automatically records detailed audit entries for operations, including who did what, when, and what data was involved. Using the Attribute [#using-the-attribute] ```csharp [Auditable] public record CreateUserCommand(string Name, string Email) : ICommand; [Auditable( IncludeRequest = true, IncludeResponse = false, ActionName = "DeleteUser", ExcludeProperties = new[] { "Password", "Token" } )] public record DeleteUserCommand(Guid UserId) : ICommand; ``` Attribute Properties [#attribute-properties] | Property | Default | Description | | ------------------- | ----------------- | ---------------------------------------- | | `IncludeRequest` | `true` | Store request payload in audit | | `IncludeResponse` | `false` | Store response payload in audit | | `ActionName` | Message type name | Custom action name | | `ExcludeProperties` | Empty | Properties to exclude from serialization | AuditEntry [#auditentry] ```csharp public class AuditEntry { public Guid Id { get; set; } public string? UserId { get; set; } public string Action { get; set; } public string EntityType { get; set; } public string? EntityId { get; set; } public string? RequestPayload { get; set; } // JSON public string? ResponsePayload { get; set; } // JSON public bool Success { get; set; } public string? ErrorMessage { get; set; } public DateTime Timestamp { get; set; } public long DurationMs { get; set; } public string? IpAddress { get; set; } public string? UserAgent { get; set; } public string? CorrelationId { get; set; } public string? Metadata { get; set; } // JSON } ``` IAuditStore [#iauditstore] ```csharp public interface IAuditStore { ValueTask SaveAsync(AuditEntry entry, CancellationToken ct); IAsyncEnumerable GetByEntityAsync( string entityType, string entityId, CancellationToken ct); IAsyncEnumerable GetByUserAsync( string userId, CancellationToken ct); IAsyncEnumerable GetByTimeRangeAsync( DateTime from, DateTime to, CancellationToken ct); } ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithAudit(options => { options.IncludeResponse = false; options.AuditFailures = true; options.ThrowOnAuditFailure = false; // Don't fail the operation if audit fails }); // Or audit everything m.WithAuditForAll(); }); ``` AuditOptions [#auditoptions] | Option | Default | Description | | --------------------- | ------- | --------------------------------- | | `IncludeResponse` | `false` | Include response in audit entries | | `AuditFailures` | `true` | Audit failed operations too | | `ThrowOnAuditFailure` | `false` | Throw if audit storage fails | Querying Audit Log [#querying-audit-log] ```csharp public class AuditController : ControllerBase { private readonly IAuditStore _auditStore; public AuditController(IAuditStore auditStore) => _auditStore = auditStore; [HttpGet("audit/user/{userId}")] public async Task GetUserAudit(string userId, CancellationToken ct) { var entries = new List(); await foreach (var entry in _auditStore.GetByUserAsync(userId, ct)) entries.Add(entry); return Ok(entries); } [HttpGet("audit/entity/{entityType}/{entityId}")] public async Task GetEntityAudit(string entityType, string entityId, CancellationToken ct) { var entries = new List(); await foreach (var entry in _auditStore.GetByEntityAsync(entityType, entityId, ct)) entries.Add(entry); return Ok(entries); } } ``` The audit behavior records both successful and failed operations (when `AuditFailures` is true), including the error message for failures. This provides a complete audit trail for compliance. # Entity Framework Core The `TurboMediator.Persistence.EntityFramework` package provides Entity Framework Core implementations for the persistence abstractions. Installation [#installation] ```bash dotnet add package TurboMediator.Persistence.EntityFramework ``` Provided Implementations [#provided-implementations] | Abstraction | EF Core Implementation | | --------------------- | -------------------------- | | `ITransactionManager` | `EfCoreTransactionManager` | | `IOutboxStore` | `EfCoreOutboxStore` | | `IInboxStore` | `EfCoreInboxStore` | | `IAuditStore` | `EfCoreAuditStore` | Registration [#registration] All at once [#all-at-once] ```csharp builder.Services.AddTurboMediator(m => { m.UseEntityFramework(); // Registers all three EF Core implementations }); // Or using the convenience method builder.Services.AddTurboMediator(m => { m.WithEntityFramework(); }); ``` Individual registration [#individual-registration] ```csharp builder.Services.AddTurboMediator(m => { m.UseEfCoreTransactions(); m.UseEfCoreOutboxStore(); m.UseEfCoreInboxStore(); m.UseEfCoreAuditStore(); }); ``` DbContext Configuration [#dbcontext-configuration] Your `DbContext` needs to include the entity configurations for outbox and audit: ```csharp public class AppDbContext : DbContext { public DbSet OutboxMessages => Set(); public DbSet InboxMessages => Set(); public DbSet AuditEntries => Set(); protected override void OnModelCreating(ModelBuilder modelBuilder) { // Apply outbox and audit entity configurations // provided by the package base.OnModelCreating(modelBuilder); } } ``` Complete Setup Example [#complete-setup-example] ```csharp var builder = WebApplication.CreateBuilder(args); // Add EF Core builder.Services.AddDbContext(options => options.UseSqlServer(builder.Configuration.GetConnectionString("Default"))); // Add TurboMediator with EF Core persistence builder.Services.AddTurboMediator(m => { m.WithEntityFramework(); m.WithTransactionForCommands(); m.WithOutbox(); m.WithInbox(); m.WithAuditForAll(); }); var app = builder.Build(); ``` Migration [#migration] After setup, create a migration to add the outbox, inbox, and audit tables: ```bash dotnet ef migrations add AddMediatorPersistence dotnet ef database update ``` # Inbox Pattern import { Callout } from 'fumadocs-ui/components/callout'; The inbox pattern complements the [outbox pattern](/docs/persistence/outbox) by providing **at-most-once processing** on the consumer side. While the outbox guarantees messages are sent, the inbox prevents duplicate processing when the same message arrives more than once. Why You Need It [#why-you-need-it] The outbox pattern provides **at-least-once delivery** — meaning messages may be delivered more than once (retries, network issues, redeliveries). The inbox pattern deduplicates these messages so your handlers process each one **exactly once**. ``` Producer Consumer ┌──────────┐ at-least-once ┌─────────┐ │ Outbox │ ─────────────────> │ Inbox │ at-most-once │ Pattern │ (may duplicate) │ Pattern │ (deduplicates) └──────────┘ └─────────┘ Together = exactly-once semantics ``` Quick Start [#quick-start] 1. Mark messages as idempotent [#1-mark-messages-as-idempotent] The simplest way is to implement `IIdempotentMessage`: ```csharp public record ProcessPaymentCommand( Guid PaymentId, decimal Amount ) : ICommand, IIdempotentMessage { public string IdempotencyKey => $"payment:{PaymentId}"; } ``` 2. Register the inbox behavior [#2-register-the-inbox-behavior] ```csharp builder.Services.AddTurboMediator(m => { m.UseEntityFramework(); m.WithInbox(); // Registers InboxBehavior pipeline }); ``` That's it. Duplicate messages with the same `IdempotencyKey` will be silently skipped. Three Ways to Define Idempotency Keys [#three-ways-to-define-idempotency-keys] Option 1: IIdempotentMessage interface (recommended) [#option-1-iidempotentmessage-interface-recommended] ```csharp public record ChargeCustomerCommand( Guid PaymentIntentId, decimal Amount, string CustomerId ) : ICommand, IIdempotentMessage { public string IdempotencyKey => $"charge:{PaymentIntentId}"; } ``` Option 2: [Idempotent] attribute with key property [#option-2-idempotent-attribute-with-key-property] ```csharp [Idempotent(KeyProperty = "OrderId")] public record ProcessOrderCommand(string OrderId, string Data) : ICommand; ``` Option 3: [Idempotent] attribute with content hash [#option-3-idempotent-attribute-with-content-hash] When no `KeyProperty` is specified, the inbox uses a SHA-256 hash of the serialized message content: ```csharp [Idempotent] public record ImportDataCommand(string Source, int BatchNumber) : ICommand; ``` Content hashing is deterministic but less explicit. Prefer `IIdempotentMessage` or `KeyProperty` for production code — they make the deduplication key obvious and debuggable. IInboxStore [#iinboxstore] ```csharp public interface IInboxStore { ValueTask HasBeenProcessedAsync( string messageId, string handlerType, CancellationToken ct); ValueTask RecordAsync(InboxMessage message, CancellationToken ct); ValueTask CleanupAsync(TimeSpan olderThan, CancellationToken ct); } ``` The store uses a composite key of **(MessageId, HandlerType)**, allowing the same message to be independently tracked across different handlers. InboxMessage [#inboxmessage] ```csharp public class InboxMessage { public string MessageId { get; set; } // Idempotency key public string HandlerType { get; set; } // Handler that processed it public string MessageType { get; set; } // Full type name public DateTime ReceivedAt { get; set; } public DateTime? ProcessedAt { get; set; } } ``` Configuration [#configuration] Basic setup [#basic-setup] ```csharp builder.Services.AddTurboMediator(m => { m.UseEntityFramework(); m.WithInbox(options => { options.RetentionPeriod = TimeSpan.FromDays(30); options.EnableAutoCleanup = true; options.CleanupInterval = TimeSpan.FromHours(6); }); }); ``` Via the outbox builder [#via-the-outbox-builder] ```csharp builder.Services.AddTurboMediator(m => { m.WithOutbox(outbox => { outbox.UseEfCoreStore() .AddProcessor() .WithInbox() // Enable inbox .WithInboxRetention(TimeSpan.FromDays(14)) // Cleanup after 14 days .UseEfCoreInboxStore(); // EF Core for inbox }); }); ``` With custom store [#with-custom-store] ```csharp builder.Services.AddTurboMediator(m => { m.WithInbox(); }); ``` InboxOptions [#inboxoptions] | Option | Default | Description | | ------------------- | ------- | ---------------------------------- | | `RetentionPeriod` | 7 days | How long inbox records are kept | | `EnableAutoCleanup` | `true` | Automatically clean up old records | | `CleanupInterval` | 1 hour | Interval between cleanup runs | How It Works [#how-it-works] ``` 1. Message arrives 2. InboxBehavior extracts idempotency key ├── IIdempotentMessage.IdempotencyKey ├── [Idempotent(KeyProperty = "...")].PropertyValue └── [Idempotent] → SHA-256 content hash 3. Check IInboxStore.HasBeenProcessedAsync(key, handlerType) ├── Already processed → return default (skip handler) └── Not processed → execute handler 4. On success → IInboxStore.RecordAsync(inboxMessage) 5. If recording fails → log warning (message was already processed successfully) ``` If inbox recording fails after the handler succeeds, the message is **not** re-processed on retry — the handler already executed. A warning is logged so you can detect this edge case. On the next delivery attempt, the record may not exist, allowing one additional execution. For strict exactly-once guarantees, use a database transaction that wraps both the handler and the inbox record. Practical Example [#practical-example] ```csharp // Webhook handler that must never process the same webhook twice public record ProcessStripeWebhookCommand( string WebhookId, string EventType, string Payload ) : ICommand, IIdempotentMessage { public string IdempotencyKey => $"stripe:{WebhookId}"; } public class ProcessStripeWebhookHandler : ICommandHandler { private readonly IPaymentService _payments; public ProcessStripeWebhookHandler(IPaymentService payments) => _payments = payments; public async ValueTask Handle( ProcessStripeWebhookCommand command, CancellationToken ct) { // Safe: inbox guarantees this runs at most once per WebhookId return command.EventType switch { "payment_intent.succeeded" => await _payments.ConfirmPaymentAsync(command.Payload, ct), "charge.refunded" => await _payments.ProcessRefundAsync(command.Payload, ct), _ => new WebhookResult(Handled: false) }; } } ``` Inbox vs Enterprise Deduplication [#inbox-vs-enterprise-deduplication] TurboMediator provides two deduplication mechanisms: | Feature | Inbox Pattern | Enterprise Deduplication | | ---------------- | --------------------------------------- | ------------------------------------------- | | Package | `TurboMediator.Persistence` | `TurboMediator.Enterprise` | | Store | `IInboxStore` (DB-backed) | `IIdempotencyStore` (in-memory/distributed) | | Use case | Persistent, transactional deduplication | Fast, cache-based deduplication | | Concurrency | Checks DB on each call | Distributed lock with wait/retry | | Response caching | No | Yes — returns cached response | | Best for | Outbox consumers, webhooks | API endpoints, concurrent request dedup | Use the **inbox pattern** when you need durable, database-backed deduplication that survives restarts — ideal for outbox consumers and webhook handlers. Use **enterprise deduplication** when you need fast, cache-based dedup with response caching — ideal for API endpoints. DbContext Setup [#dbcontext-setup] Add the `InboxMessage` entity to your DbContext: ```csharp public class AppDbContext : DbContext { public DbSet InboxMessages => Set(); protected override void OnModelCreating(ModelBuilder modelBuilder) { modelBuilder.Entity(entity => { entity.HasKey(e => new { e.MessageId, e.HandlerType }); // Composite key entity.Property(e => e.MessageId).HasMaxLength(500); entity.Property(e => e.HandlerType).HasMaxLength(500); entity.Property(e => e.MessageType).HasMaxLength(500); entity.HasIndex(e => e.ProcessedAt); // For cleanup queries }); } } ``` # Outbox Pattern import { Callout } from 'fumadocs-ui/components/callout'; The outbox pattern ensures reliable notification delivery by storing notifications in the database within the same transaction as your business data, then processing them asynchronously. How It Works [#how-it-works] 1. When a notification is published, it's serialized and stored in the outbox table (within the current transaction) 2. The transaction commits atomically — both business data and outbox messages 3. A background processor picks up pending outbox messages and publishes them to an external message broker (via `IOutboxMessageBrokerPublisher`) This ensures **at-least-once delivery** — notifications are never lost even if the application crashes after the transaction commits. IOutboxStore [#ioutboxstore] ```csharp public interface IOutboxStore { ValueTask SaveAsync(OutboxMessage message, CancellationToken ct); IAsyncEnumerable GetPendingAsync(int batchSize, CancellationToken ct); ValueTask MarkAsProcessingAsync(Guid messageId, CancellationToken ct); ValueTask TryClaimAsync(Guid messageId, string workerId, CancellationToken ct); ValueTask MarkAsProcessedAsync(Guid messageId, CancellationToken ct); ValueTask IncrementRetryAsync(Guid messageId, string error, CancellationToken ct); ValueTask MoveToDeadLetterAsync(Guid messageId, string reason, CancellationToken ct); ValueTask CleanupAsync(TimeSpan olderThan, CancellationToken ct); } ``` OutboxMessage [#outboxmessage] ```csharp public class OutboxMessage { public Guid Id { get; set; } public string MessageType { get; set; } // Full type name public string Payload { get; set; } // JSON serialized public string? CorrelationId { get; set; } public DateTime CreatedAt { get; set; } public DateTime? ProcessedAt { get; set; } public int RetryCount { get; set; } public int MaxRetries { get; set; } // Per-message retry limit public string? Error { get; set; } public OutboxMessageStatus Status { get; set; } public string? ClaimedBy { get; set; } // Worker that claimed this message public Dictionary? Headers { get; set; } } public enum OutboxMessageStatus { Pending = 0, // Awaiting processing (includes messages pending retry) Processing = 1, Processed = 2, DeadLettered = 3 // Exceeded max retries, moved to dead letter } ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithOutbox(options => { options.PublishImmediately = false; options.MaxRetries = 5; options.CorrelationIdGenerator = () => Guid.NewGuid().ToString(); }); }); ``` Using the Attribute [#using-the-attribute] Mark notifications for outbox delivery: ```csharp [WithOutbox] public record OrderCreatedNotification(Guid OrderId) : INotification; // With a specific destination (queue/topic) [WithOutbox("orders-queue")] public record OrderShippedNotification(Guid OrderId) : INotification; // With destination, partition key, and custom retry settings [WithOutbox("payments", PartitionKey = nameof(PaymentProcessedNotification.OrderId), MaxRetries = 10)] public record PaymentProcessedNotification(Guid OrderId, decimal Amount) : INotification; // Without destination but with custom settings [WithOutbox(MaxRetries = 10, PublishImmediately = true)] public record CriticalNotification(string Data) : INotification; ``` When `PublishImmediately = true`, the notification is saved to the outbox **and** published to the broker inline (if one is registered). On failure, the retry count is incremented and the message stays `Pending` so the background processor can pick it up. Practical Example [#practical-example] ```csharp [Transactional] public record PlaceOrderCommand(Guid CustomerId, List Items) : ICommand; public class PlaceOrderHandler : ICommandHandler { private readonly AppDbContext _db; private readonly IMediator _mediator; public PlaceOrderHandler(AppDbContext db, IMediator mediator) { _db = db; _mediator = mediator; } public async ValueTask Handle(PlaceOrderCommand command, CancellationToken ct) { var order = new Order { CustomerId = command.CustomerId, /* ... */ }; _db.Orders.Add(order); // This notification is stored in the outbox, not published immediately // It will be published after the transaction commits await _mediator.Publish(new OrderCreatedNotification(order.Id), ct); return order; } } ``` The outbox pattern provides **at-least-once delivery**. Your notification handlers must be **idempotent** — designed to handle the same notification being delivered more than once. See the [Inbox Pattern](/docs/persistence/inbox) for built-in deduplication on the consumer side. Message Broker Integration [#message-broker-integration] By default the outbox processor marks messages as processed without forwarding them anywhere. To actually deliver messages to an external broker (RabbitMQ, Azure Service Bus, AWS SNS/SQS, Kafka, etc.) you need to: 1. Implement `IOutboxMessageBrokerPublisher` 2. Register your implementation 3. Enable broker publishing on the processor IOutboxMessageBrokerPublisher [#ioutboxmessagebrokerpublisher] ```csharp public interface IOutboxMessageBrokerPublisher { /// Publishes to the default destination. ValueTask PublishAsync(OutboxMessage message, CancellationToken cancellationToken = default); /// Publishes to a specific destination (queue / topic). ValueTask PublishAsync(OutboxMessage message, string destination, CancellationToken cancellationToken = default); } ``` The processor calls the two-argument overload when a destination is resolved by the router (see [Message Routing](#message-routing)), and the single-argument overload otherwise. Example: RabbitMQ [#example-rabbitmq] ```csharp public class RabbitMqBrokerPublisher : IOutboxMessageBrokerPublisher { private readonly IConnection _connection; private readonly ILogger _logger; public RabbitMqBrokerPublisher(IConnection connection, ILogger logger) { _connection = connection; _logger = logger; } public ValueTask PublishAsync(OutboxMessage message, CancellationToken cancellationToken = default) => PublishAsync(message, "outbox-messages", cancellationToken); public ValueTask PublishAsync(OutboxMessage message, string destination, CancellationToken cancellationToken = default) { using var channel = _connection.CreateModel(); channel.QueueDeclare(destination, durable: true, exclusive: false, autoDelete: false); var body = Encoding.UTF8.GetBytes(message.Payload); var props = channel.CreateBasicProperties(); props.Persistent = true; props.MessageId = message.Id.ToString(); props.Type = message.MessageType; props.CorrelationId = message.CorrelationId; // Forward any headers stored on the message if (message.Headers?.Count > 0) { props.Headers = new Dictionary( message.Headers.ToDictionary(kv => kv.Key, kv => (object)kv.Value)); } channel.BasicPublish(exchange: "", routingKey: destination, basicProperties: props, body: body); _logger.LogDebug("Published outbox message {Id} to queue '{Queue}'", message.Id, destination); return ValueTask.CompletedTask; } } ``` Example: Azure Service Bus [#example-azure-service-bus] ```csharp public class ServiceBusBrokerPublisher : IOutboxMessageBrokerPublisher { private readonly ServiceBusClient _client; public ServiceBusBrokerPublisher(ServiceBusClient client) => _client = client; public ValueTask PublishAsync(OutboxMessage message, CancellationToken cancellationToken = default) => PublishAsync(message, "outbox-messages", cancellationToken); public async ValueTask PublishAsync(OutboxMessage message, string destination, CancellationToken cancellationToken = default) { var sender = _client.CreateSender(destination); var sbMessage = new ServiceBusMessage(message.Payload) { MessageId = message.Id.ToString(), CorrelationId = message.CorrelationId, ContentType = "application/json", Subject = message.MessageType, }; // Partition key for ordered delivery (set by [WithOutbox("dest", PartitionKey = ...)]) if (message.Headers?.TryGetValue("partition-key", out var partitionKey) == true) sbMessage.PartitionKey = partitionKey; if (message.Headers is not null) foreach (var (key, value) in message.Headers) sbMessage.ApplicationProperties[key] = value; await sender.SendMessageAsync(sbMessage, cancellationToken); } } ``` Registration [#registration] ```csharp builder.Services.AddTurboMediator(m => { m.WithOutbox(outbox => { outbox.AddProcessor() .PublishToMessageBroker() // enable broker mode .WithMaxRetries(5) .WithBatchSize(50); }); }); // Register your publisher implementation builder.Services.AddScoped(); // or builder.Services.AddScoped(); ``` If `PublishToMessageBroker()` is called but **no** `IOutboxMessageBrokerPublisher` is registered, the processor logs a warning and stops processing — messages remain `Pending`. Always register a publisher when broker mode is enabled. When `PublishToMessageBroker()` is **not** called (the default), the processor simply marks messages as `Processed` after they are picked up. This is useful for local development, testing, or scenarios where you consume outbox messages via direct polling instead of a broker. Message Routing [#message-routing] The router decides **which queue or topic** each message type is published to. You can specify the destination directly on the `[WithOutbox]` attribute or use the fluent builder for global configuration. Destination via [WithOutbox] [#destination-via-withoutbox] Pass the destination as the first argument and an optional `PartitionKey`: ```csharp [WithOutbox("orders")] public record OrderCreatedNotification(Guid OrderId) : INotification; // With partition key for ordered delivery (e.g. Azure Service Bus sessions) [WithOutbox("payments", PartitionKey = nameof(PaymentProcessedNotification.OrderId))] public record PaymentProcessedNotification(Guid OrderId, decimal Amount) : INotification; ``` The `PartitionKey` property names the **property on the message** whose value is read at publish time and forwarded to the broker (stored in `OutboxMessage.Headers["partition-key"]`). When no destination is specified (`[WithOutbox]`), the router falls back to naming conventions or the default destination. Fluent Routing Configuration [#fluent-routing-configuration] Use the builder to configure global routing behavior: ```csharp builder.Services.AddTurboMediator(m => { m.WithOutbox(outbox => { outbox .AddProcessor() .PublishToMessageBroker() // Auto-generate destination names from the type name .UseKebabCaseNaming() // OrderCreatedNotification → "order-created-notification" // or: .UseSnakeCaseNaming() // → "order_created_notification" // or: .UseTypeNameRouting() // → "OrderCreatedNotification" .WithDestinationPrefix("myapp.") // "myapp.order-created-notification" .WithDefaultDestination("fallback") // fallback when no rule matches // Explicit per-type overrides .MapType("orders") .MapType("payments"); }); }); ``` IOutboxMessageRouter [#ioutboxmessagerouter] For advanced scenarios you can replace the default router entirely: ```csharp public interface IOutboxMessageRouter { string GetDestination(string messageType); string GetDestination(); string GetDestination(Type type); string? GetPartitionKey(string messageType); string? GetPartitionKey(); string? GetPartitionKey(Type type); } ``` Register a custom implementation via DI: ```csharp builder.Services.AddSingleton(); ``` Routing Priority [#routing-priority] The default `OutboxMessageRouter` resolves destinations in this order: 1. **Explicit `TypeMappings`** configured via `.MapType()` or `OutboxRoutingOptions.MapType()` 2. **`[WithOutbox("destination")]`** — destination specified on the attribute 3. **Naming convention** (`KebabCase` / `SnakeCase` / `TypeName`) 4. **`DefaultDestination`** (fallback, default value: `"outbox-messages"`) Dead Letter Queue (DLQ) [#dead-letter-queue-dlq] Messages that exceed their maximum retry attempts are moved to a **dead letter** status instead of being retried indefinitely. This prevents poison messages from blocking the outbox. How It Works [#how-it-works-1] 1. `OutboxProcessor` picks up a pending message with high retry count 2. If `RetryCount >= MaxRetries`, the message is dead-lettered 3. If an `IOutboxDeadLetterHandler` is registered, it's called first (for alerts, DLQ publishing, etc.) 4. The message status changes to `DeadLettered` — even if the handler throws an exception IOutboxDeadLetterHandler [#ioutboxdeadletterhandler] ```csharp public interface IOutboxDeadLetterHandler { ValueTask HandleAsync(OutboxMessage message, string reason, CancellationToken ct); } ``` Example: Alerting on Dead Letters [#example-alerting-on-dead-letters] ```csharp public class AlertingDeadLetterHandler : IOutboxDeadLetterHandler { private readonly ILogger _logger; private readonly IAlertService _alertService; public AlertingDeadLetterHandler( ILogger logger, IAlertService alertService) { _logger = logger; _alertService = alertService; } public async ValueTask HandleAsync( OutboxMessage message, string reason, CancellationToken ct) { _logger.LogCritical( "Message {MessageId} of type {Type} dead-lettered: {Reason}", message.Id, message.MessageType, reason); await _alertService.SendAsync( $"Outbox DLQ: {message.MessageType}", $"Message {message.Id} failed after {message.RetryCount} retries. Error: {message.Error}", ct); } } ``` Registration [#registration-1] ```csharp builder.Services.AddTurboMediator(m => { m.WithOutbox(outbox => { outbox.AddProcessor() .WithMaxRetries(5) .WithDeadLetterHandler(); }); }); // Or standalone builder.Services.AddTurboMediator(m => { m.WithDeadLetterHandler(); }); ``` The dead letter handler is **optional**. Even without one, messages will still be moved to `DeadLettered` status and stop being retried. The handler allows you to take additional action (alerting, external DLQ, compensation). If the handler throws, the message is still moved to `DeadLettered` — the error is logged and appended to the reason. Multi-Worker Concurrency [#multi-worker-concurrency] The outbox processor supports running **multiple worker instances** safely — whether in the same process, multiple pods, or separate services. How It Works [#how-it-works-2] The processor uses **optimistic concurrency** via `TryClaimAsync`. When a worker picks up a batch of candidates: 1. `GetPendingAsync` returns candidate messages (a simple SELECT) 2. For each candidate, the worker calls `TryClaimAsync(messageId, workerId)` 3. `TryClaimAsync` performs an atomic `UPDATE ... WHERE Status = 'Pending' AND Id = @id` 4. If `rowsAffected == 1`, this worker won the claim and processes the message 5. If `rowsAffected == 0`, another worker already claimed it — skip This is safe for **any database** (PostgreSQL, SQL Server, MySQL, SQLite) because a single-row UPDATE is atomic at the database level. Configuration [#configuration-1] Each worker gets an auto-generated ID by default, or you can assign one explicitly: ```csharp builder.Services.AddTurboMediator(m => { m.WithOutbox(outbox => { outbox.AddProcessor() .WithProcessingInterval(TimeSpan.FromSeconds(1)) .WithBatchSize(50) .WithMaxRetries(5) .WithAutoCleanup(cleanupAge: TimeSpan.FromDays(30)); }); }); // Or configure the processor options directly builder.Services.AddOutboxProcessor(options => { options.WorkerId = "worker-pod-1"; // Explicit worker ID options.ProcessingInterval = TimeSpan.FromSeconds(1); options.BatchSize = 100; }); ``` Custom Store Implementation [#custom-store-implementation] If you implement `IOutboxStore` for a custom database, the `TryClaimAsync` implementation must be atomic: ```csharp public async ValueTask TryClaimAsync( Guid messageId, string workerId, CancellationToken ct) { // The key: UPDATE with WHERE ensures only one worker wins var sql = """ UPDATE outbox_messages SET status = 'Processing', claimed_by = @workerId WHERE id = @messageId AND status = 'Pending' """; var rowsAffected = await db.ExecuteAsync(sql, new { messageId, workerId }, ct); return rowsAffected > 0; } ``` The `ClaimedBy` column is useful for debugging: you can see which worker processed each message. It's optional for the concurrency mechanism itself — the `Status` column is what prevents duplicates. # Transactions import { Callout } from 'fumadocs-ui/components/callout'; The transaction behavior wraps handler execution in a database transaction, with automatic commit on success and rollback on failure. Installation [#installation] ```bash dotnet add package TurboMediator.Persistence dotnet add package TurboMediator.Persistence.EntityFramework # For EF Core ``` Using the Attribute [#using-the-attribute] ```csharp [Transactional] public record CreateOrderCommand(string Product, int Qty) : ICommand; [Transactional( IsolationLevel = IsolationLevel.Serializable, TimeoutSeconds = 60, AutoSaveChanges = true )] public record TransferFundsCommand( Guid FromAccount, Guid ToAccount, decimal Amount ) : ICommand; ``` Attribute Properties [#attribute-properties] | Property | Default | Description | | ----------------- | --------------- | ------------------------------- | | `IsolationLevel` | `ReadCommitted` | Transaction isolation level | | `TimeoutSeconds` | 30 | Transaction timeout | | `AutoSaveChanges` | `true` | Auto-call SaveChanges on commit | ITransactionManager [#itransactionmanager] ```csharp public interface ITransactionManager { bool HasActiveTransaction { get; } ValueTask BeginTransactionAsync( IsolationLevel isolationLevel, CancellationToken ct); ValueTask ExecuteWithStrategyAsync( Func> operation, CancellationToken ct); ValueTask SaveChangesAsync(CancellationToken ct); } ``` Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { // For specific message types m.WithTransaction(options => { options.IsolationLevel = IsolationLevel.ReadCommitted; options.AutoSaveChanges = true; options.UseExecutionStrategy = true; options.ThrowOnNestedTransaction = false; }); // Or for all commands at once m.WithTransactionForCommands(); }); ``` TransactionOptions [#transactionoptions] | Option | Default | Description | | -------------------------- | --------------- | -------------------------------------------- | | `IsolationLevel` | `ReadCommitted` | Default isolation level | | `AutoSaveChanges` | `true` | Auto-save on commit | | `UseExecutionStrategy` | `true` | Use EF Core execution strategy (for retries) | | `ThrowOnNestedTransaction` | `false` | Throw if a transaction is already active | Practical Example [#practical-example] ```csharp [Transactional] public record PlaceOrderCommand( Guid CustomerId, IReadOnlyList Items ) : ICommand; public class PlaceOrderHandler : ICommandHandler { private readonly AppDbContext _db; public PlaceOrderHandler(AppDbContext db) => _db = db; public async ValueTask Handle(PlaceOrderCommand command, CancellationToken ct) { var order = new Order { CustomerId = command.CustomerId, Items = command.Items.Select(i => new OrderLineItem { ProductId = i.ProductId, Quantity = i.Quantity, Price = i.Price }).ToList() }; _db.Orders.Add(order); // SaveChanges is called automatically by the transaction behavior // If any exception occurs, the transaction is rolled back return order; } } ``` The transaction behavior detects nested transactions and reuses the existing transaction to avoid issues. Set `ThrowOnNestedTransaction = true` if you want strict transaction boundaries. # Pipeline Behaviors import { Callout } from 'fumadocs-ui/components/callout'; Pipeline behaviors wrap the handler execution and enable cross-cutting concerns such as logging, validation, caching, retry, and more. They form a **chain of responsibility** around your handler. How the Pipeline Works [#how-the-pipeline-works] ``` Request → PreProcessor → Behavior[N] → ... → Behavior[1] → Handler → PostProcessor ↑ ExceptionHandler (on error) ``` Each behavior receives the message and a `next` delegate. Calling `next` invokes the next behavior in the chain (or the handler itself if it's the innermost behavior). IPipelineBehavior [#ipipelinebehavior] ```csharp public interface IPipelineBehavior where TMessage : IMessage { ValueTask Handle( TMessage message, MessageHandlerDelegate next, CancellationToken cancellationToken); } ``` Creating a Behavior [#creating-a-behavior] Logging Behavior [#logging-behavior] ```csharp public class LoggingBehavior : IPipelineBehavior where TMessage : IMessage { private readonly ILogger> _logger; public LoggingBehavior(ILogger> logger) { _logger = logger; } public async ValueTask Handle( TMessage message, MessageHandlerDelegate next, CancellationToken ct) { var messageType = typeof(TMessage).Name; _logger.LogInformation("Handling {MessageType}: {@Message}", messageType, message); var stopwatch = Stopwatch.StartNew(); try { var response = await next(message, ct); stopwatch.Stop(); _logger.LogInformation( "Handled {MessageType} in {ElapsedMs}ms", messageType, stopwatch.ElapsedMilliseconds); return response; } catch (Exception ex) { stopwatch.Stop(); _logger.LogError(ex, "Error handling {MessageType} after {ElapsedMs}ms", messageType, stopwatch.ElapsedMilliseconds); throw; } } } ``` Performance Behavior [#performance-behavior] ```csharp public class PerformanceBehavior : IPipelineBehavior where TMessage : IMessage { private readonly ILogger> _logger; private const int SlowThresholdMs = 500; public PerformanceBehavior(ILogger> logger) { _logger = logger; } public async ValueTask Handle( TMessage message, MessageHandlerDelegate next, CancellationToken ct) { var sw = Stopwatch.StartNew(); var response = await next(message, ct); sw.Stop(); if (sw.ElapsedMilliseconds > SlowThresholdMs) { _logger.LogWarning( "Slow handler detected: {MessageType} took {ElapsedMs}ms", typeof(TMessage).Name, sw.ElapsedMilliseconds); } return response; } } ``` Registering Behaviors [#registering-behaviors] Register behaviors using the `TurboMediatorBuilder`: ```csharp builder.Services.AddTurboMediator(m => { // Register for specific message types m.WithPipelineBehavior>(); m.WithPipelineBehavior>(); }); ``` Compile-Time Attribute Registration [#compile-time-attribute-registration] Instead of manually registering a behavior for every message type, you can use `ConfigureFromAttributes()` to let the **source generator** detect your attribute annotations at compile time and emit the DI registrations automatically — with no `Assembly.GetTypes()`, `MakeGenericType`, or reflection at runtime. Supported attributes [#supported-attributes] | Attribute | Behavior registered | | ------------------ | -------------------------------------------- | | `[RequiresTenant]` | `TenantBehavior` | | `[Authorize]` | `AuthorizationBehavior` | | `[Transactional]` | `TransactionBehavior` | | `[Auditable]` | `AuditBehavior` | Usage [#usage] Decorate your messages: ```csharp [RequiresTenant] [Authorize("CanManageProjects")] [Transactional] [Auditable(IncludeRequest = true)] public record CreateProjectCommand(string Name, string Description) : ICommand; ``` Then call `ConfigureFromAttributes()` once in your registration: ```csharp builder.Services.AddTurboMediator(m => m .ConfigureFromAttributes() .WithAuthorizationPolicies(policies => { policies.AddPolicy("CanManageProjects", u => u.IsInRole("Manager")); })); ``` The source generator scans all discovered message types at **compile time** (as part of generating the `Mediator` class) and emits a static `ConfigureFromAttributes()` extension method on `TurboMediatorBuilder` that registers only the behaviors needed by each attributed message. The generated method looks like: ```csharp // Generated inside TurboMediator.g.cs public static TurboMediatorBuilder ConfigureFromAttributes(this TurboMediatorBuilder builder) { // [RequiresTenant] on CreateProjectCommand TurboMediator.Enterprise.TurboMediatorBuilderExtensions .WithMultiTenancy(builder); // [Authorize] on CreateProjectCommand TurboMediator.Enterprise.TurboMediatorBuilderExtensions .WithAuthorization(builder); // [Transactional] on CreateProjectCommand TurboMediator.Persistence.TurboMediatorBuilderExtensions .WithTransaction(builder); // [Auditable] on CreateProjectCommand TurboMediator.Persistence.TurboMediatorBuilderExtensions .WithAudit(builder); return builder; } ``` `ConfigureFromAttributes()` is the recommended way to apply attribute-based behaviors. It is fully equivalent to calling each `WithMultiTenancy()`, `WithAuthorization()`, `WithTransaction()`, and `WithAudit()` manually — just without the boilerplate. Pipeline Execution Order [#pipeline-execution-order] Behaviors are executed in **registration order** (outermost to innermost). A typical pipeline might look like: ```csharp builder.Services.AddTurboMediator(m => { // 1. Correlation ID (outermost) m.WithCorrelationId(); // 2. Structured Logging m.WithStructuredLogging(); // 3. Telemetry m.WithTelemetry(); // 4. Authorization m.WithAuthorization(); // 5. Validation m.WithFluentValidation(); // 6. Caching m.WithCaching, object>(); // 7. Rate Limiting m.WithRateLimiting(); // 8. Transaction m.WithTransaction, object>(); // 9. Retry (innermost) m.WithRetry(); }); ``` The outermost behavior wraps all others. For example, the logging behavior should be registered first so it captures the total execution time including all other behaviors. Constrained Behaviors [#constrained-behaviors] You can constrain behaviors to specific message types using generic constraints: ```csharp // Only applies to commands public class CommandAuditBehavior : IPipelineBehavior where TCommand : IBaseCommand { public async ValueTask Handle( TCommand message, MessageHandlerDelegate next, CancellationToken ct) { // Only runs for ICommand messages var response = await next(message, ct); // Audit the command... return response; } } ``` # Exception Handlers Exception handlers allow you to intercept and handle exceptions thrown by handlers or other pipeline behaviors, without wrapping everything in try-catch blocks. IMessageExceptionHandler [#imessageexceptionhandler] ```csharp public interface IMessageExceptionHandler where TMessage : IMessage where TException : Exception { ValueTask> Handle( TMessage message, TException exception, CancellationToken cancellationToken); } ``` ExceptionHandlingResult [#exceptionhandlingresult] The handler returns one of two results: ```csharp // Exception was handled — return a fallback response ExceptionHandlingResult.HandledWith(fallbackResponse); // Exception was NOT handled — let it propagate ExceptionHandlingResult.NotHandled(); ``` Example: Fallback on Database Error [#example-fallback-on-database-error] ```csharp public class DatabaseExceptionHandler : IMessageExceptionHandler { private readonly ILogger _logger; private readonly ICacheProvider _cache; public DatabaseExceptionHandler( ILogger logger, ICacheProvider cache) { _logger = logger; _cache = cache; } public async ValueTask> Handle( GetUserQuery message, DbException exception, CancellationToken ct) { _logger.LogWarning(exception, "Database error for GetUserQuery({Id}), falling back to cache", message.Id); // Try to return cached data var cacheResult = await _cache.GetAsync($"user:{message.Id}"); if (cacheResult.HasValue) { return ExceptionHandlingResult.HandledWith(cacheResult.Value); } // Can't handle — let exception propagate return ExceptionHandlingResult.NotHandled(); } } ``` Example: Generic Error Logging [#example-generic-error-logging] ```csharp public class GenericExceptionHandler : IMessageExceptionHandler where TMessage : IMessage { private readonly ILogger> _logger; public GenericExceptionHandler( ILogger> logger) { _logger = logger; } public ValueTask> Handle( TMessage message, Exception exception, CancellationToken ct) { _logger.LogError(exception, "Unhandled exception in {MessageType}: {ErrorMessage}", typeof(TMessage).Name, exception.Message); // Let the exception propagate after logging return new ValueTask>( ExceptionHandlingResult.NotHandled()); } } ``` Registration [#registration] ```csharp builder.Services.AddTurboMediator(m => { m.WithExceptionHandler(); m.WithExceptionHandler>(); }); ``` When to Use Exception Handlers [#when-to-use-exception-handlers] * **Fallback responses** — Return cached or default data when primary source fails * **Error logging** — Centralized error logging without try-catch in every handler * **Error translation** — Convert infrastructure exceptions to domain-specific exceptions * **Graceful degradation** — Return partial results when some operations fail # Pre & Post Processors Pre-processors and post-processors provide a simpler alternative to full pipeline behaviors when you only need to run logic before or after the handler. IMessagePreProcessor [#imessagepreprocessor] Runs **before** the handler executes: ```csharp public interface IMessagePreProcessor where TMessage : IMessage { ValueTask Process(TMessage message, CancellationToken cancellationToken); } ``` Example: Validation Pre-Processor [#example-validation-pre-processor] ```csharp public class ValidationPreProcessor : IMessagePreProcessor where TMessage : IMessage { private readonly ILogger> _logger; public ValidationPreProcessor(ILogger> logger) { _logger = logger; } public ValueTask Process(TMessage message, CancellationToken ct) { _logger.LogDebug("Validating {MessageType}: {@Message}", typeof(TMessage).Name, message); // Perform custom validation logic if (message is null) throw new ArgumentNullException(nameof(message)); return default; } } ``` Example: Enrichment Pre-Processor [#example-enrichment-pre-processor] ```csharp public class TimestampPreProcessor : IMessagePreProcessor where TMessage : IMessage { private readonly IMediatorContextAccessor _contextAccessor; public TimestampPreProcessor(IMediatorContextAccessor contextAccessor) { _contextAccessor = contextAccessor; } public ValueTask Process(TMessage message, CancellationToken ct) { _contextAccessor.Context?.Items["RequestTimestamp"] = DateTime.UtcNow; return default; } } ``` IMessagePostProcessor [#imessagepostprocessor] Runs **after** the handler returns successfully: ```csharp public interface IMessagePostProcessor where TMessage : IMessage { ValueTask Process( TMessage message, TResponse response, CancellationToken cancellationToken); } ``` Example: Audit Post-Processor [#example-audit-post-processor] ```csharp public class AuditPostProcessor : IMessagePostProcessor where TMessage : IMessage { private readonly ILogger> _logger; public AuditPostProcessor(ILogger> logger) { _logger = logger; } public ValueTask Process(TMessage message, TResponse response, CancellationToken ct) { _logger.LogInformation( "Completed {MessageType} with response: {@Response}", typeof(TMessage).Name, response); return default; } } ``` Example: Cache Invalidation Post-Processor [#example-cache-invalidation-post-processor] ```csharp public class CacheInvalidationPostProcessor : IMessagePostProcessor where TCommand : IBaseCommand { private readonly ICacheProvider _cache; public CacheInvalidationPostProcessor(ICacheProvider cache) { _cache = cache; } public async ValueTask Process(TCommand message, TResponse response, CancellationToken ct) { // Invalidate related cache entries after a command var cacheKey = $"cache:{typeof(TCommand).Name}"; await _cache.RemoveAsync(cacheKey); } } ``` Registration [#registration] ```csharp builder.Services.AddTurboMediator(m => { // Register pre-processors m.WithPreProcessor>(); m.WithPreProcessor>(); // Register post-processors m.WithPostProcessor>(); m.WithPostProcessor>(); }); ``` Execution Order [#execution-order] ``` PreProcessor[1] → PreProcessor[2] → ... → Pipeline Behaviors → Handler → PostProcessor[1] → PostProcessor[2] ``` Pre-processors run in registration order before the pipeline chain. Post-processors run in registration order after the handler completes successfully. # Circuit Breaker import { Callout } from 'fumadocs-ui/components/callout'; The circuit breaker monitors failures and temporarily stops sending requests to a failing service, giving it time to recover. States [#states] ``` Closed ──(failures exceed threshold)──→ Open ──(after cooldown)──→ HalfOpen ↑ │ └────────(success in half-open)──────────────────────────────────────┘ │ Open ←──(failure in half-open)───┘ ``` | State | Behavior | | ------------ | ----------------------------------------------------------------- | | **Closed** | Normal operation. Failures are counted. | | **Open** | All requests fail immediately with `CircuitBreakerOpenException`. | | **HalfOpen** | A limited number of test requests are allowed through. | Configuration [#configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithCircuitBreaker(options => { options.FailureThreshold = 5; // Open after 5 failures options.OpenDuration = TimeSpan.FromSeconds(30); // Stay open for 30s options.SuccessThreshold = 1; // Close after 1 success in half-open }); }); ``` CircuitBreakerOptions [#circuitbreakeroptions] | Option | Default | Description | | ------------------ | ---------- | -------------------------------------- | | `FailureThreshold` | 5 | Number of failures before opening | | `OpenDuration` | 30 seconds | How long the circuit stays open | | `SuccessThreshold` | 1 | Successes needed in half-open to close | Inspecting State [#inspecting-state] The circuit breaker exposes static methods for inspection and control: ```csharp // Get the current state var state = CircuitBreakerBehavior.GetCircuitState(); // Returns: Closed, Open, or HalfOpen // Manually reset a circuit CircuitBreakerBehavior.Reset(); // Reset all circuits CircuitBreakerBehavior.ResetAll(); ``` Practical Example [#practical-example] ```csharp // Configuration builder.Services.AddTurboMediator(m => { m.WithCircuitBreaker(opt => { opt.FailureThreshold = 3; opt.OpenDuration = TimeSpan.FromSeconds(60); }); }); // Message public record ProcessPaymentCommand( Guid OrderId, decimal Amount, string Currency ) : ICommand; // Handler public class ProcessPaymentHandler : ICommandHandler { private readonly IPaymentGateway _gateway; public ProcessPaymentHandler(IPaymentGateway gateway) => _gateway = gateway; public async ValueTask Handle( ProcessPaymentCommand command, CancellationToken ct) { return await _gateway.ChargeAsync( command.OrderId, command.Amount, command.Currency, ct); } } // Usage — after 3 failures, the circuit opens and // subsequent calls fail immediately without hitting the gateway try { var result = await mediator.Send(new ProcessPaymentCommand(orderId, 99.99m, "USD")); } catch (CircuitBreakerOpenException) { // Circuit is open — return a friendly error return Results.StatusCode(503); } ``` Circuit state is maintained in a static `ConcurrentDictionary` keyed by message type. This means the circuit is shared across all instances of the same message type within the same application process. # Fallback import { Callout } from 'fumadocs-ui/components/callout'; The fallback behavior provides alternative responses when the primary handler fails. You can chain multiple fallback handlers with priority ordering and exception type filtering. IFallbackHandler [#ifallbackhandler] ```csharp public interface IFallbackHandler where TMessage : IMessage { ValueTask HandleFallbackAsync( TMessage message, Exception exception, CancellationToken cancellationToken); } ``` Using the Attribute [#using-the-attribute] ```csharp [Fallback(typeof(CachedProductFallback), Order = 1)] [Fallback(typeof(DefaultProductFallback), Order = 2)] public record GetProductQuery(Guid Id) : IQuery; ``` Attribute Properties [#attribute-properties] | Property | Description | | --------------------- | ----------------------------------------- | | `typeof(HandlerType)` | The fallback handler type | | `Order` | Execution priority (lower runs first) | | `OnExceptionTypes` | Only trigger for specific exception types | Fallback Handlers [#fallback-handlers] ```csharp // Primary fallback: try cache public class CachedProductFallback : IFallbackHandler { private readonly ICacheProvider _cache; public CachedProductFallback(ICacheProvider cache) => _cache = cache; public async ValueTask HandleFallbackAsync( GetProductQuery message, Exception exception, CancellationToken ct) { var cached = await _cache.GetAsync($"product:{message.Id}"); return cached ?? throw new InvalidOperationException("No cached product"); } } // Secondary fallback: return default public class DefaultProductFallback : IFallbackHandler { public ValueTask HandleFallbackAsync( GetProductQuery message, Exception exception, CancellationToken ct) { return new ValueTask(Product.NotFound(message.Id)); } } ``` Using Configuration [#using-configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithFallback(options => { options.ThrowOnAllFallbacksFailure = true; options.OnFallbackInvoked = (info) => { Console.WriteLine($"Fallback {info.FallbackHandlerType.Name} invoked for {info.MessageType} (attempt {info.AttemptNumber})"); }; }); }); ``` FallbackOptions [#fallbackoptions] | Option | Default | Description | | ---------------------------- | ------- | ------------------------------------------- | | `ExceptionTypes` | All | Exception types that trigger fallback | | `ShouldHandle` | `null` | Custom predicate for when to use fallback | | `ThrowOnAllFallbacksFailure` | `true` | Throw if all fallbacks fail | | `OnFallbackInvoked` | `null` | Callback when a fallback is used | | `DefaultValueFactory` | `null` | Factory for default value if all else fails | Exception-Specific Fallbacks [#exception-specific-fallbacks] ```csharp [Fallback(typeof(NetworkFallback), OnExceptionTypes = new[] { typeof(HttpRequestException) })] [Fallback(typeof(TimeoutFallback), OnExceptionTypes = new[] { typeof(TimeoutException) })] [Fallback(typeof(GenericFallback), Order = 99)] public record GetDataQuery(string Key) : IQuery; ``` Fallbacks are tried in `Order` sequence. The first fallback that returns successfully wins. If a fallback also throws, the next fallback in the chain is tried. # Hedging import { Callout } from 'fumadocs-ui/components/callout'; Hedging sends **multiple parallel requests** with staggered delays, using the first successful response and cancelling the rest. This technique reduces tail latency by racing requests against each other. How It Works [#how-it-works] 1. Sends the initial request 2. After a configurable delay, sends another parallel request 3. Repeats up to `MaxParallelAttempts` 4. Returns the **first successful** response 5. Cancels all remaining in-flight requests Using the Attribute [#using-the-attribute] ```csharp [Hedging(maxParallelAttempts: 3, DelayMs = 200)] public record GetPriceQuery(string Symbol) : IQuery; ``` Using Configuration [#using-configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithHedging(options => { options.MaxParallelAttempts = 3; options.Delay = TimeSpan.FromMilliseconds(200); options.CancelPendingOnSuccess = true; }); }); ``` HedgingOptions [#hedgingoptions] | Option | Default | Description | | ------------------------ | ------- | ----------------------------------- | | `MaxParallelAttempts` | 2 | Max parallel requests | | `Delay` | 100ms | Delay before launching next attempt | | `CancelPendingOnSuccess` | `true` | Cancel remaining on first success | Practical Example [#practical-example] ```csharp // Race multiple attempts to get stock price [Hedging(maxParallelAttempts: 3, DelayMs = 150)] public record GetStockPriceQuery(string Symbol) : IQuery; public class GetStockPriceHandler : IQueryHandler { private readonly IStockApi _api; public GetStockPriceHandler(IStockApi api) => _api = api; public async ValueTask Handle( GetStockPriceQuery query, CancellationToken ct) { // If this takes too long, a parallel attempt will race it return await _api.GetPriceAsync(query.Symbol, ct); } } ``` Hedging increases load on downstream services because it sends multiple parallel requests. Use it wisely for read operations that are idempotent. **Do not use hedging for write operations** that could cause duplicates. When to Use Hedging [#when-to-use-hedging] * **High-availability reads** — Reduce tail latency for time-sensitive queries * **Multi-region deployments** — Race requests to different regions * **Unreliable external APIs** — Hedge against slow responses When NOT to Use [#when-not-to-use] * **Write operations** — Could cause unintended duplicates * **High-cost operations** — Each attempt consumes resources * **Rate-limited APIs** — Multiple attempts consume rate limit budget # Retry import { Callout } from 'fumadocs-ui/components/callout'; The retry behavior automatically retries failed operations with configurable exponential backoff and jitter, making your application resilient to transient failures. Installation [#installation] ```bash dotnet add package TurboMediator.Resilience ``` Basic Usage [#basic-usage] Using Attributes [#using-attributes] The simplest way to add retry is via the `[Retry]` attribute: ```csharp [Retry(maxAttempts: 3, delayMilliseconds: 1000)] public record SendEmailCommand(string To, string Subject, string Body) : ICommand; ``` Using Configuration [#using-configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithRetry(options => { options.MaxAttempts = 3; options.DelayMilliseconds = 1000; options.UseExponentialBackoff = true; options.MaxDelayMilliseconds = 30000; }); }); ``` Attribute Options [#attribute-options] ```csharp [Retry( maxAttempts: 5, delayMilliseconds: 500, UseExponentialBackoff = true, MaxDelayMilliseconds = 15000, RetryOnExceptions = new[] { typeof(HttpRequestException), typeof(TimeoutException) } )] public record CallExternalApiQuery(string Endpoint) : IQuery; ``` | Parameter | Default | Description | | ----------------------- | ------- | ------------------------------------ | | `maxAttempts` | 3 | Maximum number of retry attempts | | `delayMilliseconds` | 1000 | Base delay between retries | | `UseExponentialBackoff` | `true` | Doubles delay on each retry | | `MaxDelayMilliseconds` | 30000 | Maximum delay cap | | `RetryOnExceptions` | All | Exception types that trigger a retry | RetryOptions [#retryoptions] ```csharp public class RetryOptions { public int MaxAttempts { get; set; } = 3; public int DelayMilliseconds { get; set; } = 1000; public bool UseExponentialBackoff { get; set; } = true; public int MaxDelayMilliseconds { get; set; } = 30000; public Type[]? RetryOnExceptions { get; set; } } ``` How It Works [#how-it-works] 1. Executes the handler 2. If the handler throws an exception matching `RetryOnExceptions`: * Waits for `delay + random jitter` * Retries the handler (up to `MaxAttempts`) 3. If exponential backoff is enabled, each retry doubles the delay 4. If all attempts fail, the last exception is thrown Practical Example [#practical-example] ```csharp // Message [Retry(maxAttempts: 3, delayMilliseconds: 200, UseExponentialBackoff = true)] public record SyncInventoryCommand(string ProductId) : ICommand; // Handler public class SyncInventoryHandler : ICommandHandler { private readonly IExternalInventoryApi _api; public SyncInventoryHandler(IExternalInventoryApi api) => _api = api; public async ValueTask Handle( SyncInventoryCommand command, CancellationToken ct) { // This may fail due to network issues — retry will handle it var stock = await _api.GetStockLevel(command.ProductId, ct); return new SyncResult(command.ProductId, stock); } } ``` The retry behavior adds random jitter to prevent the **thundering herd** problem when multiple services retry simultaneously. # Timeout The timeout behavior cancels handler execution if it exceeds a configured duration, preventing slow operations from blocking indefinitely. Using the Attribute [#using-the-attribute] ```csharp [Timeout(5000)] // 5 seconds public record GenerateReportQuery(DateRange Range) : IQuery; ``` Using Configuration [#using-configuration] ```csharp builder.Services.AddTurboMediator(m => { m.WithTimeout(TimeSpan.FromSeconds(5)); }); ``` How It Works [#how-it-works] The behavior races the handler execution against a `Task.Delay`: 1. Starts the handler and a delay timer in parallel 2. If the handler completes first → returns the result 3. If the timer completes first → throws `TimeoutException` Practical Example [#practical-example] ```csharp // Slow external API call with 3-second timeout [Timeout(3000)] public record FetchWeatherQuery(string City) : IQuery; public class FetchWeatherHandler : IQueryHandler { private readonly HttpClient _httpClient; public FetchWeatherHandler(HttpClient httpClient) => _httpClient = httpClient; public async ValueTask Handle( FetchWeatherQuery query, CancellationToken ct) { var response = await _httpClient.GetFromJsonAsync( $"/api/weather/{query.City}", ct); return response!; } } // Usage try { var weather = await mediator.Send(new FetchWeatherQuery("London")); } catch (TimeoutException) { // Handler took too long return Results.StatusCode(504); } ``` Combining with Retry [#combining-with-retry] Timeout works well with retry — each attempt has its own timeout: ```csharp [Retry(maxAttempts: 3, delayMilliseconds: 500)] [Timeout(2000)] public record CallApiCommand(string Endpoint) : ICommand; ``` # Entity Framework Core import { Callout } from 'fumadocs-ui/components/callout'; Installation [#installation] ```bash dotnet add package TurboMediator.Scheduling.EntityFramework ``` Quick Setup [#quick-setup] ```csharp builder.Services.AddDbContext(options => options.UseSqlServer(builder.Configuration.GetConnectionString("Default"))); builder.Services.AddTurboMediator(m => m .WithScheduling(scheduling => { scheduling.UseEfCoreStore(); scheduling.AddRecurringJob("cleanup") .WithCron("0 3 * * *") .WithRetry(RetryStrategy.Fixed(3, 60)) .WithData(() => new CleanupCommand()); }) ); ``` SchedulingDbContext [#schedulingdbcontext] The `SchedulingDbContext` provides two tables: ```csharp public class SchedulingDbContext : DbContext { public DbSet RecurringJobs => Set(); public DbSet JobOccurrences => Set(); } ``` Table: SchedulingRecurringJobs [#table-schedulingrecurringjobs] | Column | Type | Notes | | -------------------------- | ---------------- | ------------------------ | | `JobId` | `varchar(256)` | Primary key | | `MessageTypeName` | `nvarchar(max)` | | | `CronExpression` | `nvarchar(max)` | Nullable | | `IntervalTicks` | `bigint` | TimeSpan stored as ticks | | `NextRunAt` | `datetimeoffset` | Nullable | | `LastRunAt` | `datetimeoffset` | Nullable | | `Status` | `nvarchar(max)` | String conversion | | `RetryIntervalSecondsJson` | `nvarchar(max)` | JSON array | | `SkipIfAlreadyRunning` | `bit` | | | `Priority` | `nvarchar(max)` | String conversion | | `MessagePayload` | `nvarchar(max)` | Nullable | | `TimeZoneId` | `nvarchar(max)` | Nullable | | `RowVersion` | `rowversion` | Optimistic concurrency | **Index:** `IX_SchedulingRecurringJobs_Status_NextRunAt` on (`Status`, `NextRunAt`) Table: SchedulingJobOccurrences [#table-schedulingjoboccurrences] | Column | Type | Notes | | -------------------- | ------------------ | ----------------- | | `Id` | `uniqueidentifier` | Primary key | | `JobId` | `nvarchar(256)` | | | `StartedAt` | `datetimeoffset` | | | `CompletedAt` | `datetimeoffset` | Nullable | | `Status` | `nvarchar(max)` | String conversion | | `RetryCount` | `int` | | | `Error` | `nvarchar(max)` | Nullable | | `ParentOccurrenceId` | `uniqueidentifier` | Nullable | | `RunCondition` | `nvarchar(max)` | String conversion | **Index:** `IX_SchedulingJobOccurrences_JobId_StartedAt` on (`JobId`, `StartedAt`) Migrations [#migrations] Generate migrations using the standard EF Core tooling: ```bash dotnet ef migrations add InitScheduling --context SchedulingDbContext dotnet ef database update --context SchedulingDbContext ``` If you're using a separate migration assembly: ```csharp builder.Services.AddDbContext(options => options.UseSqlServer(connectionString, sql => sql.MigrationsAssembly("MyApp.Migrations"))); ``` Optimistic Concurrency [#optimistic-concurrency] The `EfCoreJobStore.TryLockJobAsync` uses EF Core's `RowVersion` concurrency token: 1. Load the job 2. Set `Status = Running` 3. Call `SaveChangesAsync` 4. If `DbUpdateConcurrencyException` is thrown, another instance locked it first → return `false` This ensures safe multi-instance deployments without distributed locks. ``` Instance A: Load job → Status=Running → Save ✓ (locked) Instance B: Load job → Status=Running → Save ✗ (concurrency conflict) ``` Custom DbContext Configuration [#custom-dbcontext-configuration] You can extend or customize the `SchedulingDbContext`: ```csharp builder.Services.AddDbContext(options => options.UseNpgsql(connectionString) // PostgreSQL .EnableDetailedErrors() .EnableSensitiveDataLogging()); ``` The SchedulingDbContext is independent from your application's DbContext. Make sure to run migrations separately for the scheduling context. Supported Databases [#supported-databases] Any EF Core provider works. Tested with: * SQL Server * PostgreSQL (Npgsql) * SQLite * In-Memory (for testing) ```csharp // Testing with InMemory builder.Services.AddDbContext(options => options.UseInMemoryDatabase("SchedulingTests")); ``` # Scheduling import { Callout } from 'fumadocs-ui/components/callout'; `TurboMediator.Scheduling` adds native **cron jobs and recurring job scheduling** to TurboMediator. The core idea: **a job is an `ICommand`. An `ICommand` has a pipeline. Scheduling is just defining when `IMediator.Send()` gets called.** Installation [#installation] ```bash dotnet add package TurboMediator.Scheduling ``` For EF Core persistence: ```bash dotnet add package TurboMediator.Scheduling.EntityFramework ``` Key Features [#key-features] * **Cron expressions** (5-field and 6-field with seconds) via [Cronos](https://github.com/HangfireIO/Cronos) * **Interval-based** scheduling (Every, Hourly, Daily, Weekly) * **Retry with custom intervals** per attempt * **SkipIfAlreadyRunning** — anti-overlap protection * **Pause / Resume / TriggerNow** runtime controls * **Job occurrence history** — full execution log * **Priority hints** including `LongRunning` (dedicated thread) * **IJobExecutionContext** — rich context injected into handlers * **SkipJobException / TerminateJobException** — control flow exceptions * **Timezone support** for cron evaluation * **Zero reflection** — typed dispatch via source-generator-friendly registry Quick Start [#quick-start] ```csharp // Define a command (same as any other TurboMediator command) public record CleanupExpiredTokensCommand() : ICommand; public class CleanupHandler : ICommandHandler { public ValueTask Handle(CleanupExpiredTokensCommand command, CancellationToken ct) { // cleanup logic... return Unit.ValueTask; } } // Register in Program.cs builder.Services.AddTurboMediator(m => m .WithScheduling(scheduling => { scheduling.UseInMemoryStore(); scheduling.AddRecurringJob("cleanup-tokens") .WithCron("0 3 * * *") // daily at 03:00 UTC .WithRetry(RetryStrategy.Fixed(2, 60)) .WithData(() => new CleanupExpiredTokensCommand()); }) ); ``` That's it. The `RecurringJobProcessor` background service starts automatically and dispatches due jobs through the full mediator pipeline. Your handlers don't need to know they're being called by a scheduler. They receive the same command, go through the same pipeline behaviors (validation, resilience, observability), and return the same response. Architecture [#architecture] ``` ┌─────────────────────────────────────┐ │ SchedulingBuilder │ ← Fluent config at startup │ AddRecurringJob / WithCron / Every │ └──────────────┬──────────────────────┘ │ seeds ▼ ┌─────────────────────────────────────┐ │ IJobStore │ ← InMemory or EF Core │ RecurringJobRecord + Occurrences │ └──────────────┬──────────────────────┘ │ polls ▼ ┌─────────────────────────────────────┐ │ RecurringJobProcessor │ ← BackgroundService │ (PeriodicTimer, default 10s) │ └──────────────┬──────────────────────┘ │ dispatches (zero reflection) ▼ ┌─────────────────────────────────────┐ │ IMediator.Send(command) │ ← Full pipeline │ Validation → Resilience → Handler │ └─────────────────────────────────────┘ ``` Runtime Management [#runtime-management] Inject `IJobScheduler` to manage jobs at runtime: ```csharp app.MapPost("/api/jobs/{jobId}/pause", async (string jobId, IJobScheduler scheduler) => { await scheduler.PauseJobAsync(jobId); return Results.Ok(); }); app.MapPost("/api/jobs/{jobId}/trigger", async (string jobId, IJobScheduler scheduler) => { await scheduler.TriggerNowAsync(jobId); return Results.Accepted(); }); // Query execution history var history = await scheduler.GetOccurrencesAsync("cleanup-tokens", limit: 20); ``` # Job Chaining vs Saga import { Callout } from 'fumadocs-ui/components/callout'; Overview [#overview] TurboMediator provides two powerful coordination mechanisms that may seem similar at first glance but serve fundamentally different purposes: | Aspect | Scheduling | Saga | | ---------------- | -------------------------- | ------------------------------------- | | **Purpose** | *When* to run | *What* to run after what | | **Trigger** | Time-based (cron/interval) | Event-based (previous step completes) | | **State** | Stateless between runs | Stateful with persistent context | | **Compensation** | No rollback | Built-in compensation/rollback | | **Scope** | Background tasks | Business workflows | | **Concurrency** | SkipIfAlreadyRunning | Step-level idempotency | Scheduling: Temporal Triggers [#scheduling-temporal-triggers] The scheduling system answers: **"When should this task run?"** ```csharp // Run cleanup every night at 3 AM scheduling.AddRecurringJob("nightly-cleanup") .WithCron("0 3 * * *") .WithRetry(RetryStrategy.Fixed(3, 60)) .WithData(() => new CleanupCommand()); // Sync external data every 5 minutes scheduling.AddRecurringJob("data-sync") .Every(TimeSpan.FromMinutes(5)) .SkipIfAlreadyRunning() .WithData(() => new SyncCommand()); ``` Characteristics: * Each execution is **independent** — no shared state between runs * Retry is about recovering from transient errors in the *same* task * No concept of "if step A succeeds, do step B" * No compensation or rollback Saga: Workflow Orchestration [#saga-workflow-orchestration] The saga system answers: **"What should happen next, and what if it fails?"** ```csharp public class OrderSaga : Saga { public override void Configure(ISagaBuilder builder) { builder .Start() .Then() .WithCompensation() .Then() .WithCompensation() .Then(); } } ``` Characteristics: * Steps are **causally linked** — each depends on the previous * **Persistent state** flows through the entire workflow * **Compensation** automatically rolls back on failure * Triggered by an event, not a clock Why Not Job Chaining in Scheduling? [#why-not-job-chaining-in-scheduling] Early design considered adding Job Chaining to the scheduling module: ``` Job A → (OnSuccess) → Job B → (OnFailure) → Job C ``` We decided **against** this because: 1. **Overlap with Saga**: Chaining jobs based on success/failure is workflow orchestration — exactly what Sagas do with better primitives (state, compensation, retries per step). 2. **No shared state**: Chained scheduled jobs would have no way to pass data between steps without inventing a state mechanism — which is just reinventing Sagas. 3. **No compensation**: If Job B fails after Job A succeeds, there's no built-in way to undo Job A's effects. 4. **Simpler model**: Keeping scheduling focused on temporal concerns makes it easier to reason about and test. The `RunCondition` enum and `ParentOccurrenceId` field exist in the data model as reserved fields. They are not used by the processor today but allow potential future extensions without schema changes. When to Use What [#when-to-use-what] Use Scheduling when: [#use-scheduling-when] * You need to run a task on a **time-based** schedule (cron, interval) * Each execution is **independent** and stateless * Retry is needed for transient failures only * Examples: cleanup, health checks, data sync, report generation, cache warming Use Saga when: [#use-saga-when] * You have a **multi-step business workflow** * Steps must execute in order with **data flowing** between them * You need **compensation** if a step fails * The workflow is triggered by a **business event**, not a clock * Examples: order processing, payment flows, user onboarding, provisioning Combine Both: [#combine-both] The most powerful pattern is **combining scheduling with sagas**: ```csharp // Handler triggered by scheduler public class BatchProcessHandler : ICommandHandler { private readonly ISender _sender; public BatchProcessHandler(ISender sender) => _sender = sender; public async ValueTask Handle(BatchProcessCommand cmd, CancellationToken ct) { var pendingOrders = await GetPendingOrders(); foreach (var order in pendingOrders) { // Start a saga for each pending order await _sender.Send(new StartOrderSagaCommand(order.Id), ct); } return Unit.Value; } } // Registration scheduling.AddRecurringJob("process-pending-orders") .WithCron("0 */30 * * *") // every 30 minutes .WithData(() => new BatchProcessCommand()); ``` In this pattern: * **Scheduling** handles the *when* (every 30 minutes) * **Saga** handles the *how* (multi-step order processing with compensation) Summary [#summary] ``` ┌──────────────────────────────────────────┐ │ TurboMediator Coordination │ ├────────────────────┬─────────────────────┤ │ Scheduling │ Saga │ ├────────────────────┼─────────────────────┤ │ Cron / Interval │ Event-triggered │ │ Stateless │ Stateful │ │ Retry (transient) │ Compensation │ │ Independent runs │ Linked steps │ │ "Run X at Y time" │ "After A, do B" │ └────────────────────┴─────────────────────┘ ``` Think of it this way: **Scheduling is a clock. Saga is a recipe.** The clock tells you when to start cooking; the recipe tells you each step and what to do if you burn the sauce. # Job Store import { Callout } from 'fumadocs-ui/components/callout'; IJobStore [#ijobstore] The `IJobStore` interface defines how job definitions and occurrence records are persisted: ```csharp public interface IJobStore { // Job definitions Task UpsertJobAsync(RecurringJobRecord job, CancellationToken ct = default); Task GetJobAsync(string jobId, CancellationToken ct = default); Task> GetAllJobsAsync(CancellationToken ct = default); Task> GetDueJobsAsync(DateTimeOffset now, CancellationToken ct = default); Task RemoveJobAsync(string jobId, CancellationToken ct = default); // Locking (concurrency) Task TryLockJobAsync(string jobId, CancellationToken ct = default); Task ReleaseJobAsync(string jobId, CancellationToken ct = default); // Occurrence tracking Task AddOccurrenceAsync(JobOccurrenceRecord occurrence, CancellationToken ct = default); Task UpdateOccurrenceAsync(JobOccurrenceRecord occurrence, CancellationToken ct = default); Task> GetOccurrencesAsync( string jobId, int limit = 20, CancellationToken ct = default); } ``` Built-in: In-Memory Store [#built-in-in-memory-store] For development and testing, a fast in-memory store is provided by default: ```csharp scheduling.UseInMemoryStore(); ``` * Uses `ConcurrentDictionary` and `SemaphoreSlim` for thread safety * Data is lost on application restart * Locking is process-local (not suitable for multi-instance deployments) RecurringJobRecord [#recurringjobrecord] The job definition model: | Property | Type | Description | | ---------------------- | ----------------- | ------------------------------------------------- | | `JobId` | `string` | Unique identifier | | `MessageTypeName` | `string` | Full type name of the command | | `CronExpression` | `string?` | Cron expression (null if interval-based) | | `Interval` | `TimeSpan?` | Fixed interval (null if cron-based) | | `NextRunAt` | `DateTimeOffset?` | Next scheduled execution | | `LastRunAt` | `DateTimeOffset?` | Last completed execution | | `Status` | `JobStatus` | Current status (Scheduled, Running, Paused, etc.) | | `RetryIntervalSeconds` | `int[]` | Retry intervals derived from RetryStrategy | | `SkipIfAlreadyRunning` | `bool` | Whether to skip overlapping executions | | `Priority` | `JobPriority` | Execution priority | | `MessagePayload` | `string?` | Serialized command data (JSON) | | `TimeZoneId` | `string?` | IANA timezone identifier | JobOccurrenceRecord [#joboccurrencerecord] Each execution of a recurring job creates an occurrence: | Property | Type | Description | | -------------------- | ----------------- | ------------------------------------------ | | `Id` | `Guid` | Unique occurrence ID | | `JobId` | `string` | Parent job identifier | | `StartedAt` | `DateTimeOffset` | When execution began | | `CompletedAt` | `DateTimeOffset?` | When execution finished | | `Status` | `JobStatus` | Outcome (Done, Failed, Skipped, Cancelled) | | `RetryCount` | `int` | Number of retries performed | | `Error` | `string?` | Error message if failed | | `ParentOccurrenceId` | `Guid?` | Reserved for future chaining | | `RunCondition` | `RunCondition` | Reserved for future chaining | IJobScheduler (Runtime Management) [#ijobscheduler-runtime-management] The `IJobScheduler` interface allows runtime control of jobs: ```csharp public interface IJobScheduler { Task PauseJobAsync(string jobId, CancellationToken ct = default); Task ResumeJobAsync(string jobId, CancellationToken ct = default); Task RemoveJobAsync(string jobId, CancellationToken ct = default); Task TriggerNowAsync(string jobId, CancellationToken ct = default); Task> GetOccurrencesAsync( string jobId, int limit = 20, CancellationToken ct = default); Task GetJobAsync(string jobId, CancellationToken ct = default); Task> GetAllJobsAsync(CancellationToken ct = default); } ``` Exposing via API Endpoints [#exposing-via-api-endpoints] ```csharp var jobs = app.MapGroup("/api/jobs"); jobs.MapGet("/", async (IJobScheduler scheduler) => Results.Ok(await scheduler.GetAllJobsAsync())); jobs.MapPost("/{id}/pause", async (string id, IJobScheduler scheduler) => { await scheduler.PauseJobAsync(id); return Results.NoContent(); }); jobs.MapPost("/{id}/resume", async (string id, IJobScheduler scheduler) => { await scheduler.ResumeJobAsync(id); return Results.NoContent(); }); jobs.MapPost("/{id}/trigger", async (string id, IJobScheduler scheduler) => { await scheduler.TriggerNowAsync(id); return Results.Accepted(); }); jobs.MapDelete("/{id}", async (string id, IJobScheduler scheduler) => { await scheduler.RemoveJobAsync(id); return Results.NoContent(); }); jobs.MapGet("/{id}/occurrences", async (string id, IJobScheduler scheduler) => Results.Ok(await scheduler.GetOccurrencesAsync(id))); ``` Custom Job Store [#custom-job-store] Implement `IJobStore` for any persistence backend: ```csharp public class MongoJobStore : IJobStore { private readonly IMongoCollection _jobs; private readonly IMongoCollection _occurrences; // Implement all IJobStore methods... } ``` Register it: ```csharp scheduling.UseStore(); ``` For production use with Entity Framework Core, see the [Entity Framework](./entity-framework) page which provides a ready-made `IJobStore` implementation with optimistic concurrency. Polling Configuration [#polling-configuration] ```csharp scheduling.Configure(options => { options.PollingInterval = TimeSpan.FromSeconds(5); // default: 10s }); ``` The processor uses `PeriodicTimer` for accurate, non-drifting intervals. Shorter polling intervals mean lower latency but higher database load. # Recurring Jobs import { Callout } from 'fumadocs-ui/components/callout'; Cron Expression [#cron-expression] Schedule jobs using standard 5-field or 6-field (with seconds) cron expressions: ```csharp scheduling.AddRecurringJob("my-job") .WithCron("0 3 * * *") // daily at 03:00 .WithData(() => new MyCommand()); // With seconds (6-field) scheduling.AddRecurringJob("ping") .WithCron("*/30 * * * * *") // every 30 seconds .WithData(() => new PingCommand()); ``` Cron parsing is powered by [Cronos](https://github.com/HangfireIO/Cronos) (MIT, zero transitive dependencies). Interval-Based [#interval-based] For simpler schedules, use fixed intervals: ```csharp scheduling.AddRecurringJob("health-check") .Every(TimeSpan.FromMinutes(5)) .WithData(() => new HealthCheckCommand()); // Convenience methods .Hourly() // TimeSpan.FromHours(1) .Daily() // TimeSpan.FromDays(1) .Weekly() // TimeSpan.FromDays(7) ``` Timezone Support [#timezone-support] By default, cron expressions are evaluated in UTC. To use a specific timezone: ```csharp scheduling.AddRecurringJob("daily-report") .WithCron("0 8 * * 1-5") // 08:00 on weekdays .WithTimeZone(TimeZoneInfo.FindSystemTimeZoneById("America/Sao_Paulo")) .WithData(() => new DailyReportCommand()); ``` SkipIfAlreadyRunning [#skipifalreadyrunning] Prevent overlap when a job takes longer than its schedule interval: ```csharp scheduling.AddRecurringJob("long-process") .WithCron("0 * * * *") // every hour .SkipIfAlreadyRunning() // skip if previous run still active .WithData(() => new LongProcessCommand()); ``` When skipped, a `JobOccurrenceRecord` with `Status = Skipped` is recorded. Priority [#priority] ```csharp scheduling.AddRecurringJob("rebuild-index") .WithCron("0 2 * * 0") // Sundays at 02:00 .WithPriority(JobPriority.LongRunning) .WithData(() => new RebuildIndexCommand()); ``` | Priority | Behavior | | ------------- | -------------------------------------------------------------------- | | `LongRunning` | Dispatched with `TaskCreationOptions.LongRunning` (dedicated thread) | | `High` | Metadata for ordering (processed first) | | `Normal` | Default | | `Low` | Metadata for ordering (processed last) | [RecurringJob] Attribute [#recurringjob-attribute] As an alternative to the fluent builder, you can annotate a handler class directly with `[RecurringJob]`: ```csharp [RecurringJob("nightly-cleanup", "0 3 * * *")] public class CleanupHandler : ICommandHandler { public ValueTask Handle(CleanupCommand cmd, CancellationToken ct) { // ... return Unit.ValueTask; } } ``` The attribute signature: ```csharp [AttributeUsage(AttributeTargets.Class, AllowMultiple = false, Inherited = false)] public sealed class RecurringJobAttribute : Attribute { public string JobId { get; } public string CronExpression { get; } public RecurringJobAttribute(string jobId, string cronExpression) { ... } } ``` Auto-Registration via Source Generator [#auto-registration-via-source-generator] The source generator detects `[RecurringJob]` at compile time and generates a `ConfigureRecurringJobsFromAttributes()` extension method. Call it inside `WithScheduling` to auto-register all attributed handlers — **zero reflection**: ```csharp builder.Services.AddTurboMediator(m => m .WithScheduling(scheduling => { scheduling.UseInMemoryStore(); // Auto-registers all [RecurringJob] handlers at compile time scheduling.ConfigureRecurringJobsFromAttributes(); // You can still add builder-based jobs alongside attribute-based ones scheduling.AddRecurringJob("report") .WithCron("0 8 * * 1-5") .WithRetry(RetryStrategy.Fixed(3, 60)) .WithData(() => new ReportCommand()); }) ); ``` The command type must have a **parameterless constructor** for attribute-based registration (the source generator generates `new TCommand()` as the default payload). Attribute vs Builder [#attribute-vs-builder] | Feature | `[RecurringJob]` | Builder | | -------------------- | -------------------------- | ----------------------------- | | Cron schedule | Yes | Yes | | Interval schedule | No | Yes (`Every`, `Hourly`, etc.) | | Retry strategy | No (default: none) | Yes (`WithRetry`) | | SkipIfAlreadyRunning | No (default: false) | Yes | | Priority | No (default: Normal) | Yes (`WithPriority`) | | Timezone | No (default: UTC) | Yes (`WithTimeZone`) | | Custom payload | No (uses `new TCommand()`) | Yes (`WithData`) | Use the attribute for simple cron jobs that need no extra configuration. Use the builder for full control. IJobExecutionContext [#ijobexecutioncontext] Handlers can inject `IJobExecutionContext` to access execution metadata: ```csharp public class ImportHandler : ICommandHandler { private readonly IJobExecutionContext _ctx; public ImportHandler(IJobExecutionContext ctx, ILogger logger) { _ctx = ctx; } public ValueTask Handle(ImportCommand cmd, CancellationToken ct) { Console.WriteLine($"Job: {_ctx.JobId}, Attempt: {_ctx.RetryCount + 1}"); if (_ctx.RetryCount >= 2) { // Fallback strategy on 3rd attempt } return Unit.ValueTask; } } ``` Properties available: | Property | Description | | -------------------- | --------------------------------------------- | | `JobId` | Unique job identifier | | `OccurrenceId` | ID of this specific execution | | `RetryCount` | Number of retries so far (0 on first attempt) | | `StartedAt` | When the occurrence started | | `CronExpression` | The cron expression that triggered this run | | `ParentOccurrenceId` | Parent ID if this is a chained child job | SkipJobException [#skipjobexception] Throw from a handler to skip the current occurrence without retry: ```csharp throw new SkipJobException("Data already imported today"); ``` The occurrence is recorded with `Status = Skipped`. TerminateJobException [#terminatejobexception] Terminate execution with a specific status, bypassing retry: ```csharp // Terminate as failed (no retry) throw new TerminateJobException(JobStatus.Failed, "Invalid configuration"); // Terminate as cancelled throw new TerminateJobException(JobStatus.Cancelled, "Feature disabled"); ``` Regular exceptions (not Skip/Terminate) will trigger the retry strategy. If all retries are exhausted, the occurrence is recorded as `Failed`. # Retry Strategies import { Callout } from 'fumadocs-ui/components/callout'; Overview [#overview] When a handler throws an exception (other than `SkipJobException` or `TerminateJobException`), the scheduling processor retries according to the configured `RetryStrategy`. The retry intervals are stored as an `int[]` of seconds on the `RecurringJobRecord`. Built-in Strategies [#built-in-strategies] None (Default) [#none-default] No retries. The occurrence is immediately marked as `Failed`. ```csharp .WithRetry(RetryStrategy.None) ``` Fixed [#fixed] Retries a fixed number of times with a constant delay: ```csharp // 3 retries, each after 60 seconds .WithRetry(RetryStrategy.Fixed(attempts: 3, delaySeconds: 60)) ``` Produces intervals: `[60, 60, 60]` Exponential Backoff [#exponential-backoff] Retries with exponentially increasing delays: ```csharp // 5 retries: 2s, 4s, 8s, 16s, 32s .WithRetry(RetryStrategy.ExponentialBackoff(maxAttempts: 5, baseSeconds: 2)) ``` Produces intervals: `[2, 4, 8, 16, 32]` Immediate [#immediate] Retries immediately (0-second delay): ```csharp // 3 immediate retries .WithRetry(RetryStrategy.Immediate(attempts: 3)) ``` Produces intervals: `[0, 0, 0]` Custom [#custom] Provide exact intervals manually: ```csharp // Wait 10s, then 30s, then 120s .WithRetry(RetryStrategy.Custom(new[] { 10, 30, 120 })) ``` How Retries Work [#how-retries-work] ``` Job fails → Check retry intervals ├── RetryCount < intervals.Length → Wait intervals[RetryCount] seconds → Retry └── RetryCount >= intervals.Length → Mark as Failed (exhausted) ``` 1. On first failure, `RetryCount = 0`, the processor waits `intervals[0]` seconds 2. On second failure, `RetryCount = 1`, it waits `intervals[1]` seconds 3. This continues until all intervals are exhausted 4. Each retry creates a new loop iteration with updated `RetryCount` on the `IJobExecutionContext` Retry vs Skip vs Terminate [#retry-vs-skip-vs-terminate] | Scenario | What to use | Retries? | | --------------------------------------- | -------------------------------------- | -------------------------- | | Transient failure (network, timeout) | Let the exception propagate | Yes | | Business logic skip (already processed) | `throw new SkipJobException(...)` | No | | Fatal error (bad config) | `throw new TerminateJobException(...)` | No | | Expected failure pattern | `RetryStrategy.Custom(...)` | Yes, with custom intervals | Accessing Retry Count in Handlers [#accessing-retry-count-in-handlers] ```csharp public class ResilientHandler : ICommandHandler { private readonly IJobExecutionContext _ctx; private readonly HttpClient _http; public ResilientHandler(IJobExecutionContext ctx, HttpClient http) { _ctx = ctx; _http = http; } public async ValueTask Handle(ApiSyncCommand cmd, CancellationToken ct) { if (_ctx.RetryCount > 0) { // Use fallback endpoint on retry await _http.GetAsync("/api/fallback/sync", ct); } else { await _http.GetAsync("/api/primary/sync", ct); } return Unit.Value; } } ``` Complete Example [#complete-example] ```csharp builder.Services.AddTurboMediator(m => m .WithScheduling(scheduling => { scheduling.UseInMemoryStore(); scheduling.AddRecurringJob("sync-orders") .WithCron("0 */2 * * *") // every 2 hours .WithRetry(RetryStrategy.ExponentialBackoff( maxAttempts: 4, baseSeconds: 5)) // 5s, 10s, 20s, 40s .SkipIfAlreadyRunning() .WithData(() => new SyncOrdersCommand()); }) ); ``` Retry intervals are persisted in the job store as `RetryIntervalSeconds` (an `int[]`). This means you can inspect and even modify them at runtime through a custom `IJobStore` implementation. # CLI Tool import { Callout } from 'fumadocs-ui/components/callout'; The `dotnet-turbo` CLI tool provides analysis, documentation generation, health monitoring, and benchmarking capabilities. Installation [#installation] ```bash dotnet tool install --global TurboMediator.Cli ``` Commands [#commands] analyze — Handler Coverage [#analyze--handler-coverage] Analyzes your project for handler coverage, identifying missing or duplicate handlers: ```bash dotnet-turbo analyze -p ./src/MyApp ``` Output: ``` 📊 TurboMediator Analysis ───────────────────────── Messages found: 15 Handlers found: 14 ✅ CreateUserCommand → CreateUserHandler ✅ GetUserQuery → GetUserHandler ✅ UserCreatedNotification → EmailHandler, LogHandler ⚠️ OrphanedCommand → No handler found! Coverage: 93% (14/15) ``` docs — Generate Documentation [#docs--generate-documentation] Generates API documentation from your codebase in markdown, HTML, or JSON format: ```bash # Markdown (default) dotnet-turbo docs -p ./src/MyApp -o ./docs -f markdown # HTML dotnet-turbo docs -p ./src/MyApp -o ./docs/api -f html # JSON (for programmatic use) dotnet-turbo docs -p ./src/MyApp -o ./api-docs.json -f json ``` Generated documentation includes: * All message types with their properties * All handlers with their associated messages * Pipeline behaviors and their order * XML doc comments (if present) health — Health Check [#health--health-check] Check the health of a running TurboMediator application: ```bash # One-time check dotnet-turbo health -e https://localhost:5001/health # Watch mode with live table dotnet-turbo health -e https://localhost:5001/health --watch ``` Watch mode displays a live `Spectre.Console` table that auto-refreshes. benchmark — Performance Analysis [#benchmark--performance-analysis] Analyzes handler complexity and estimates performance characteristics: ```bash dotnet-turbo benchmark -p ./src/MyApp ``` Output: ``` 🚀 TurboMediator Benchmarks ─────────────────────────── Handler Analysis: CreateUserHandler Complexity: Medium Async Operations: 2 (DB write, notification publish) Estimated Latency: 5-15ms GetUserHandler Complexity: Low Async Operations: 1 (DB read) Estimated Latency: 1-5ms ProcessVideoHandler Complexity: High Async Operations: 4 (API call, S3 upload, DB write, notification) Estimated Latency: 100-500ms ``` The CLI tool uses MSBuild/Roslyn workspace to analyze your source code at the syntax and semantic level. It doesn't require compiling or running your application. # Testing import { Callout } from 'fumadocs-ui/components/callout'; The testing package provides comprehensive tools for unit testing handlers, behaviors, and mediator interactions. Installation [#installation] ```bash dotnet add package TurboMediator.Testing ``` FakeMediator [#fakemediator] An in-memory fake `IMediator` for isolated unit tests: ```csharp var mediator = new FakeMediator(); // Setup responses mediator.Setup(cmd => new User(Guid.NewGuid(), cmd.Name, cmd.Email)); mediator.SetupQuery(query => new User(query.Id, "Test User", "test@example.com")); // Setup exceptions mediator.SetupException( new InvalidOperationException("User not found")); // Use in tests var user = await mediator.Send(new CreateUserCommand("Alice", "alice@example.com")); Assert.Equal("Alice", user.Name); ``` Verification [#verification] ```csharp // Verify a command was sent mediator.Verify( cmd => cmd.Name == "Alice", Times.Once()); // Verify a query was sent mediator.VerifyQuery( q => q.Id == expectedId, Times.Once()); // Verify a notification was published mediator.VerifyPublished( n => n.Email == "alice@example.com", Times.Once()); // Verify exact count mediator.Verify(Times.Exactly(2)); // Verify never sent mediator.Verify(Times.Never()); ``` Inspection [#inspection] ```csharp // Get all sent messages of a type var commands = mediator.GetSentMessages(); Assert.Single(commands); // Get all published notifications var notifications = mediator.GetPublishedNotifications(); Assert.Equal(2, notifications.Count); ``` Times [#times] ```csharp Times.Never() // Exactly 0 Times.Once() // Exactly 1 Times.Exactly(3) // Exactly 3 Times.AtLeastOnce() // 1 or more Times.AtLeast(2) // 2 or more Times.AtMost(5) // 5 or fewer Times.Between(2, 5) // Between 2 and 5 ``` RecordingMediator [#recordingmediator] Wraps a **real mediator** and records all interactions: ```csharp var services = new ServiceCollection(); services.AddTurboMediator(); var provider = services.BuildServiceProvider(); var realMediator = provider.GetRequiredService(); var recording = new RecordingMediator(realMediator); // Use normally await recording.Send(new CreateUserCommand("Alice", "alice@example.com")); await recording.Publish(new UserCreatedNotification(userId, "alice@example.com")); // Inspect records Assert.Equal(1, recording.Commands.Count()); Assert.Equal(1, recording.Notifications.Count()); // Detailed record inspection var record = recording.Records.First(); Assert.Equal(MessageKind.Command, record.MessageKind); Assert.True(record.IsSuccess); Assert.True(record.Duration < TimeSpan.FromSeconds(1)); ``` MessageRecord [#messagerecord] ```csharp public class MessageRecord { public object Message { get; } public MessageKind MessageKind { get; } public DateTime SentAt { get; } public DateTime? CompletedAt { get; } public object? Response { get; } public Exception? Exception { get; } public bool IsSuccess { get; } public TimeSpan? Duration { get; } } public enum MessageKind { Command, Query, Request, Notification, StreamRequest, StreamCommand, StreamQuery } ``` Filtering Extensions [#filtering-extensions] ```csharp var failedCommands = recording.Records .WhereMessage() .Failed() .ToList(); var slowQueries = recording.Records .WhereMessage(q => q.Id == specificId) .Successful() .Where(r => r.Duration > TimeSpan.FromSeconds(1)) .ToList(); ``` Handler Test Base Classes [#handler-test-base-classes] Abstract base classes for isolated handler testing: Request Handler Test [#request-handler-test] ```csharp public class PingHandlerTests : RequestHandlerTestBase { protected override PingHandler CreateHandler() { return new PingHandler(); } [Fact] public async Task Should_Return_Pong() { var result = await Handle(new PingRequest()); Assert.Equal("Pong!", result); } } ``` Command Handler Test [#command-handler-test] ```csharp public class CreateUserHandlerTests : CommandHandlerTestBase { private readonly Mock _repoMock = new(); protected override CreateUserHandler CreateHandler() { return new CreateUserHandler(_repoMock.Object); } [Fact] public async Task Should_Create_User() { var result = await Handle(new CreateUserCommand("Alice", "alice@example.com")); Assert.Equal("Alice", result.Name); _repoMock.Verify(r => r.AddAsync(It.IsAny(), It.IsAny())); } } ``` Query Handler Test [#query-handler-test] ```csharp public class GetUserHandlerTests : QueryHandlerTestBase { private readonly Mock _repoMock = new(); protected override GetUserHandler CreateHandler() { return new GetUserHandler(_repoMock.Object); } [Fact] public async Task Should_Return_User() { var userId = Guid.NewGuid(); _repoMock.Setup(r => r.FindByIdAsync(userId, It.IsAny())) .ReturnsAsync(new User(userId, "Alice", "alice@example.com")); var result = await Handle(new GetUserQuery(userId)); Assert.NotNull(result); Assert.Equal("Alice", result!.Name); } } ``` Pipeline Behavior Test [#pipeline-behavior-test] ```csharp public class LoggingBehaviorTests : PipelineBehaviorTestBase< LoggingBehavior, PingRequest, string> { private readonly Mock>> _loggerMock = new(); protected override LoggingBehavior CreateBehavior() { return new LoggingBehavior(_loggerMock.Object); } [Fact] public async Task Should_Log_And_Call_Next() { var result = await InvokePipeline(new PingRequest(), "Pong!"); Assert.Equal("Pong!", result); // Verify logging happened } [Fact] public async Task Should_Log_Exceptions() { await Assert.ThrowsAsync(() => InvokePipelineWithException( new PingRequest(), new InvalidOperationException("test error"))); } } ``` TestScenario (Given/When/Then) [#testscenario-givenwhenthen] BDD-style test builder: ```csharp [Fact] public async Task Scenario_Create_And_Get_User() { var userId = Guid.NewGuid(); await TestScenario.Create() .Given(mediator => { mediator.Setup(cmd => new User(userId, cmd.Name, cmd.Email)); mediator.SetupQuery(q => new User(q.Id, "Alice", "alice@example.com")); }) .When(async mediator => { var user = await mediator.Send( new CreateUserCommand("Alice", "alice@example.com")); await mediator.Publish( new UserCreatedNotification(user.Id, user.Email)); }) .ThenVerify(Times.Once()) .ThenVerifyPublished(Times.Once()) .Execute(); } ``` MediatorTestFixture [#mediatortestfixture] For integration tests with real DI: ```csharp public class IntegrationTests { [Fact] public async Task Full_Pipeline_Test() { var fixture = new MediatorTestFixture(); fixture.AddSingleton(new InMemoryUserRepository()); var mediator = fixture.Mediator; var result = await mediator.Send(new CreateUserCommand("Alice", "alice@example.com")); Assert.NotNull(result); } } ``` `FakeMediator` is ideal for unit tests where you control all dependencies. `RecordingMediator` is better for integration tests where you want to verify interactions with a real pipeline.