Skip to main content

Azure Cosmos DB Provider

The Cosmos DB provider implements ICloudNativePersistenceProvider for globally distributed, multi-model database access with partition key management, tunable consistency, and change feed subscriptions.

Before You Start​

  • .NET 8.0+ (or .NET 9/10 for latest features)
  • An Azure Cosmos DB account and database
  • Familiarity with data access and IDb interface

Installation​

dotnet add package Excalibur.Data.CosmosDb

Dependencies: Excalibur.Data.Abstractions, Microsoft.Azure.Cosmos

Quick Start​

using Microsoft.Extensions.DependencyInjection;

services.AddCosmosDb(options =>
{
    options.AccountEndpoint = "https://myaccount.documents.azure.com:443/";
    options.AccountKey = "<your-auth-key>";
    options.DatabaseName = "MyDatabase";
});

Registration Methods​

MethodWhat It RegistersKey Options
AddCosmosDb(opts)Core persistence providerAccountEndpoint, AccountKey, DatabaseName
AddCosmosDbSnapshotStore(opts)ISnapshotStoreContainerName
AddCosmosDbProjectionStore<T>(connStr, dbName, opts?)IProjectionStore<T>ContainerName

All methods also accept IConfiguration binding: AddCosmosDb(configuration, sectionName: "CosmosDb").

Batch Projection Registration​

Register multiple projections sharing the same Cosmos DB account:

services.AddCosmosDbProjections(connectionString, "MyDatabase", projections =>
{
    projections.Add<OrderSummary>();
    projections.Add<CustomerProfile>(o => o.ContainerName = "customers");
});

Change Data Capture​

services.AddCdcProcessor(cdc =>
{
    cdc.UseCosmosDb(options =>
    {
        options.DatabaseName = "MyApp";
        options.ContainerName = "orders";
        options.LeaseContainerName = "leases";
    })
    .TrackTable("orders", t => t.MapAll<OrderChangedEvent>())
    .EnableBackgroundProcessing();
});

Partition Keys​

Cosmos DB requires partition keys for data distribution:

// Simple partition key
var key = new PartitionKey("tenant-123", "/tenantId");
var order = await provider.GetByIdAsync<Order>("order-1", key, consistencyOptions: null, cancellationToken: ct);

// Composite partition key
var compositeKey = new CompositePartitionKey(
    path: "/pk",
    separator: "|",
    "region-us", "tenant-123");

Consistency Options​

Tune consistency per operation:

// Strong consistency
var result = await provider.QueryAsync<Order>(
    "SELECT * FROM c WHERE c.status = 'pending'",
    partitionKey,
    parameters: null,
    consistencyOptions: ConsistencyOptions.Strong,
    cancellationToken: ct);

// Session consistency (read-your-own-writes)
var sessionOptions = ConsistencyOptions.WithSession(sessionToken);

// Bounded staleness
var boundedOptions = ConsistencyOptions.WithBoundedStaleness(
    maxStaleness: TimeSpan.FromSeconds(5),
    maxVersionLag: 100);

Change Feed​

Subscribe to real-time document changes:

var subscription = await provider.CreateChangeFeedSubscriptionAsync<Order>(
    containerName: "orders",
    options: new ChangeFeedOptions
    {
        StartPosition = ChangeFeedStartPosition.Beginning,
        MaxBatchSize = 100,
        PollingInterval = TimeSpan.FromSeconds(5)
    },
    cancellationToken: ct);

await subscription.StartAsync(ct);

await foreach (var change in subscription.ReadChangesAsync(ct))
{
    // Process change event
}

Batch Operations​

Execute multiple operations in a single transaction within a partition:

var result = await provider.ExecuteBatchAsync(
    partitionKey,
    new ICloudBatchOperation[]
    {
        // batch operations
    },
    cancellationToken: ct);

if (result.Success)
{
    Console.WriteLine($"Request charge: {result.RequestCharge} RU");
}

ETag-Based Concurrency​

UpdateAsync accepts an optional ETag for optimistic concurrency. It returns a CloudOperationResult<TDocument> with conflict detection:

// Update with ETag check (etag obtained from a previous write or Cosmos DB SDK)
var updateResult = await provider.UpdateAsync(
    modifiedOrder,
    partitionKey,
    etag: previousEtag,
    cancellationToken: ct);

if (updateResult.IsConcurrencyConflict)
{
    // Handle conflict — another process modified the document (HTTP 412)
}

// Successful updates return the new ETag for subsequent operations
var newEtag = updateResult.ETag;

See Also​