NoSQL in .NET: A Practical Guide to Azure Cosmos DB and MongoDB

// Reserve two units atomically, only if enough stock is available
await products.PatchItemAsync<Product>(
    id: productId,
    partitionKey: new PartitionKey(categoryId),
    patchOperations:
    [
        PatchOperation.Increment("/stock/available", -2),
        PatchOperation.Increment("/stock/reserved", 2),
        PatchOperation.Set("/updatedAt", DateTimeOffset.UtcNow)
    ],
    requestOptions: new PatchItemRequestOptions
    {
        FilterPredicate = "FROM p WHERE p.stock.available >= 2" // else 412
    },
    cancellationToken: ct);

var pk = new PartitionKey(order.CustomerId);
var message = new OutboxMessage(
    Id: Guid.NewGuid().ToString(), CustomerId: order.CustomerId, Type: "OrderPlaced",
    OrderId: order.Id);

using TransactionalBatchResponse result = await orders
    .CreateTransactionalBatch(pk)
    .CreateItem(order)
    .CreateItem(message)
    .PatchItem($"summary-{order.CustomerId}", [PatchOperation.Increment("/orderCount", 1)])
    .ExecuteAsync(ct);

if (!result.IsSuccessStatusCode)
{
    // All or nothing: no operation was applied. 409 means a duplicate id, for example.
    throw new InvalidOperationException($"Order batch failed with {result.StatusCode}.");
}

var bulkClient = new CosmosClient(endpoint, new DefaultAzureCredential(),
    new CosmosClientOptions { AllowBulkExecution = true });
var target = bulkClient.GetContainer("shop", "products");

// Bulk mode is not transactional: capture failures per item instead of failing the run
var results = await Task.WhenAll(batch.Select(async product =>
{
    try
    {
        var r = await target.UpsertItemAsync(
            product, new PartitionKey(product.CategoryId), cancellationToken: ct);
        return (product.Id, Charge: r.RequestCharge, Error: (string?)null);
    }
    catch (CosmosException ex)
    {
        return (product.Id, Charge: ex.RequestCharge, Error: ex.StatusCode.ToString());
    }
}));

var failed = results.Where(r => r.Error is not null).ToList();
Console.WriteLine($"{results.Length - failed.Count} upserted, {failed.Count} failed, " +
    $"{results.Sum(r => r.Charge):N0} RU");

public sealed class OrderProjectionService(
    CosmosClient client, IReadModelStore readModels, ILogger<OrderProjectionService> logger)
    : IHostedService
{
    private ChangeFeedProcessor? _processor;

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var orders = client.GetContainer("shop", "orders");
        var leases = client.GetContainer("shop", "leases"); // partitioned on /id

        _processor = orders
            .GetChangeFeedProcessorBuilder<Order>("order-projections", HandleChangesAsync)
            .WithInstanceName(Environment.MachineName) // unique per running instance
            .WithLeaseContainer(leases)
            .Build();

        await _processor.StartAsync();
    }

    public Task StopAsync(CancellationToken cancellationToken) =>
        _processor?.StopAsync() ?? Task.CompletedTask;

    private async Task HandleChangesAsync(
        ChangeFeedProcessorContext context, IReadOnlyCollection<Order> changes,
        CancellationToken ct)
    {
        foreach (var order in changes)
        {
            // Delivery is at least once, so every handler must be idempotent
            await readModels.UpsertOrderSummaryAsync(order, ct);
        }

        logger.LogInformation("Lease {Lease}: {Count} changes, {Charge} RU",
            context.LeaseToken, changes.Count, context.Headers.RequestCharge);
    }
}