Corner49.Infra.AzureCosmosDB 10.0.169

Corner49.Infra.AzureCosmosDB

Opinionated repository layer for Azure Cosmos DB (NoSQL API). Part of Corner49.Infra – a set of packages that provide opinionated tooling for infrastructure in .NET applications.

• One DocumentRepo<T> per container, with point reads, LINQ and SQL queries, paging, streaming, patching, bulk operations and change feed
• Databases and containers are created at startup when they do not exist
• One shared CosmosClient per connection string
• System.Text.Json serialisation with camelCase names
• Built-in retry on 429 / 408, and RU pressure monitoring for heavy jobs

dotnet add package Corner49.Infra.AzureCosmosDB

Already included when you install Corner49.Infra.

Configuration

{
  "CosmosDB": {
    "ConnectString": "AccountEndpoint=https://...;AccountKey=...;",
    "DatabaseName": "data",
    "DirectMode": false
  }
}

Getting started

1. Define a repository

using Corner49.Infra.DB;

public interface IOrderRepo {
    Task<Order?> Get(string customerId, string id);
    Task<QueryResult<Order>> Open(string customerId, string? continuationToken);
    Task Save(Order order);
}

public class OrderRepo : IOrderRepo, IDocumentRepoInitializer {
    private readonly DocumentRepo<Order> _repo;

    public OrderRepo(IDocumentDB db) {
        // container name, then the partition key path(s)
        _repo = db.GetRepo<Order>("Orders", "customerId");
    }

    Task IDocumentRepoInitializer.Init() => _repo.Init();

    public Task<Order?> Get(string customerId, string id) => _repo.GetItem(customerId, id);

    public Task<QueryResult<Order>> Open(string customerId, string? continuationToken) =>
        _repo.Query(customerId, q => q.Where(o => o.Status == "open"), continuationToken, maxItemCount: 50);

    public Task Save(Order order) => _repo.UpsertItem(order.CustomerId, order);
}

IDocumentRepoInitializer.Init() is called for every registered repo at startup; DocumentRepo<T>.Init() creates the database and container when they are missing.

2. Register it

With an infra builder (Corner49.Infra or Corner49.Infra.CLI):

infra.AddDocumentDB(db => {
    db.Configure = cfg => cfg.DatabaseName = "shop";   // optional override of the configuration
    db.AddRepo<IOrderRepo, OrderRepo>();
});

With a plain IServiceCollection:

var db = services.AddDocumentDB(configuration, b => b.AddRepo<IOrderRepo, OrderRepo>());
// after the host is built:
await db.Init(app.Services);

Or without dependency injection:

var db = new DocumentDB(connectString, dbName: "shop");
var repo = db.GetRepo<Order>("Orders", "customerId");
await repo.Init();

Repos are registered as singletons. Always go through IDocumentDB rather than creating your own CosmosClient.

Partition keys

GetRepo<T>(container, params string[] partitionKey) takes one path for a simple key or several for a hierarchical key. Every operation has a string overload for the first case and a string[] overload for the second:

_repo = db.GetRepo<Order>("Orders", "tenantId", "customerId");
await _repo.GetItem(new[] { tenantId, customerId }, id);

Pass null as the partition key to query across partitions. That is a fan-out query: use it deliberately.

DocumentRepo<T> API

Continuation tokens returned by Query and Filter are opaque strings: pass them back unchanged for the next page. Always use @param placeholders with the parameters dictionary instead of concatenating values into SQL.

Throughput

By default Init() creates the database and container without explicit throughput. Set autoscale limits on the repo before it is initialised:

_repo = db.GetRepo<Order>("Orders", "customerId");
_repo.ContainerAutoscaleThroughput = 4000;   // autoscale max RU/s

Point operations are retried on 429 TooManyRequests and 408 RequestTimeout, honouring the server's retry-after.

For RU-heavy loops, pause before you get throttled instead of retrying afterwards:

foreach (var order in orders) {
    await _repo.WaitForCapacity();                // waits while usage is above 80% of provisioned RU/s
    await _repo.UpsertItem(order.CustomerId, order);
}

GetThroughputStats() returns the observed RU/s, the provisioned RU/s, their ratio and the number of recent 429s over a rolling window. The numbers only cover requests made through that repo instance in the current process.

Diagnostics

_repo.OnDiagnostics = diag => {
    logger.LogInformation("{Method} {Ms} ms, {RU} RU", diag.Method, diag.ElapsedTime?.TotalMilliseconds, diag.TotalRequestCharge);
    return Task.CompletedTask;
};

Errors

Failures surface as DocumentException (with a StatusCode). DocumentContainerNotFoundException means the container was used before Init() ran. A missing document is not an error: reads return null and deletes return false.

Serialisation

Documents are serialised with System.Text.Json using JsonHelper.Options from Corner49.Infra.Core: camelCase property names, enums as strings, nulls omitted. Use [JsonPropertyName] only where the stored name must differ, and make sure your document exposes id.

Related packages

• Corner49.Infra – web apps and APIs
• Corner49.Infra.CLI – console apps and jobs
• Corner49.Infra.Core – abstractions and utilities

Source: https://github.com/FrankVDL2471/Corner49.Infra