Winche.Database.AspNetCore 5.2.0

Winche.Database

A JSON document database layer built on top of PostgreSQL. Store, query, and subscribe to JSON documents — with PostgreSQL as the storage backend via JSONB.

Supports real-time live queries with indexed delta listeners, optimistic ACID transactions, aggregation pipelines, durable document lifecycle hooks, filtered secondary indexes, and integrates with Winche.Sentinel for per-document access control.

Protocol and migration docs:

• PROTOCOL — complete v3 wire format reference (values, writes, queries, pipelines, REST API, WebSocket protocol)
• RELEASE-NOTES-v3 — breaking changes and behavior changes from v2

Packages

Install

dotnet add package Winche.Database
dotnet add package Winche.Database.AspNetCore
dotnet add package Winche.Database.AspNetCore.Rest
dotnet add package Winche.Database.AspNetCore.WebSockets

Add only the transport packages you need (Rest, WebSockets, or both).

Quick Start

1. Configure appsettings.json

{
  "ConnectionStrings": {
    "WincheDatabase": "Host=localhost;Database=mydb;Username=postgres;Password=secret"
  }
}

The library does not read IConfiguration itself — you pass the connection string (and any store options) explicitly in AddWincheDatabase (step 2). Where you keep them is your choice.

All database objects (table, changes feed, cursors, winche_* functions, indexes) are created unqualified and resolve through the connection's schema search path — to host a store in a non-public schema, set it in the connection string (Search Path=myschema) rather than in code.

2. Register services

using Winche.Database.DependencyInjection;
using Winche.Database.AspNetCore.DependencyInjection;
using Winche.Database.AspNetCore.Rest.DependencyInjection;
using Winche.Database.AspNetCore.WebSockets.DependencyInjection;

builder.Services
    .AddWincheDatabase(config =>
    {
        // Connection (required); use "Search Path=myschema" for non-public schemas
        config.ConnectionString = builder.Configuration.GetConnectionString("WincheDatabase")!;

        // Store behavior (optional; these are the defaults)
        config.TransactionConfig = new() { IdleTimeoutSpan = TimeSpan.FromMinutes(1) };

        // Access rules (evaluated on every protected operation)
        config.AddDocumentAccessRule<MyReadRule>();
        config.AddDocumentAccessRule<MyWriteRule>();

        // Document lifecycle hooks (at-least-once, feed-driven, must be idempotent)
        config.AddDocumentStoreHook<MyHook>();

        // Secondary index definitions
        config.AddIndexDefinition<MyIndexDefinition>();

        // Claims accessor — reads auth context from the HTTP request or WebSocket connection
        config.SetCallerClaimsAccessor<MyClaimsAccessor>();
    })
    .AddWincheDatabaseWsApi();   // WebSocket transport

SetCallerClaimsAccessor is provided by Winche.Database.AspNetCore.DependencyInjection and registers the accessor for both the REST and WebSocket transports in one call.

3. Initialize schema and map routes

await app.InitializeWincheDatabaseAsync();   // Creates winche_* tables/functions and syncs indexes
app.UseWebSockets();                         // required before mapping the WS endpoint
app.MapWincheDatabaseWsApi();                // WebSocket endpoint: /documents/ws
app.MapWincheDatabaseRestApi();              // REST routes under /documents (configurable)

Both Map* methods return an IEndpointConventionBuilder covering all of their endpoints — for REST that includes the CRUD routes and every colon-verb (:commit, :runQuery, :aggregate, …); for WS the upgrade route. Apply cross-cutting policy on it once and it lands everywhere:

app.MapWincheDatabaseRestApi().RequireAuthorization();   // CRUD AND :commit/:runQuery/:aggregate/…
app.MapWincheDatabaseWsApi().RequireRateLimiting("ws");  // throttle socket upgrades

The built-in claims/exception filters are always applied internally and run outermost, so caller conventions compose on top of them.

WebSocket auth is in-band, not at the upgrade. A WS connection authenticates via the token in its hello frame, validated by IWsAuthenticator (this exists because browsers can't set request headers on the upgrade). Do not rely on .RequireAuthorization() to secure WS — it gates the HTTP upgrade against HttpContext.User and is independent of the hello token. Use it only as an optional defense-in-depth gate, and only if you have a scheme that authenticates the upgrade request itself (cookies, or a query-string token — not a bearer header).

Features

• typed engine — 11 tagged value types (null, bool, int64, double, timestamp, string, bytes, reference, geopoint, array, map); cross-type total order; same-type-class inequality semantics; __name__ tiebreaker; int/double numeric equality
• Document storage — Store documents as typed field maps; each document carries path, id, collection, createTime, updateTime, and version metadata
• Querying — 15 filter operators (including Winche extensions: arrayContainsAll, contains, startsWith, endsWith, regex, field-compare); and/or/not composites; orderBy with __name__ tiebreaker; cursor-based pagination (startAt/startAfter/endAt/endBefore)
• Live queries with delta listeners — Subscribe to a query and receive an initial full snapshot, then indexed added/modified/removed deltas with count checksum; listen.snapshot REPLACES client state, listen.delta MUTATES it
• Optimistic transactions — Optimistic read-version ledger; ABORTED on conflict with safe retry; RunTransactionAsync with automatic retry; reads-before-writes enforced; idle (60 s) and absolute (5 min) timeouts
• Batch writes — Atomic commit of up to 500 set/update/delete operations with field transforms, preconditions, and a single commit timestamp; use new WriteBatch(db) for a fluent builder
• Field transforms — serverTimestamp, increment (saturating int, promotes to double on mixed), maximum, minimum, arrayUnion, arrayRemove
• Aggregation pipelines — Multi-stage pipelines: match, filter, lookup, unwind, group (with having), project, sort, limit, skip; 9 accumulators: count, sum, avg, min, max, push, addToSet, first, last
• Durable hooks — Feed-driven, true at-least-once end-to-end; hooks execute inline+sequentially per batch; failed batches retried with capped backoff; hooks fire for writes from any node; cursor persisted across restarts; idempotency required
• Secondary indexes — composite expression indexes per collection; IndexDefinition.Where for filtered/partial indexes (agreement tested against engine); wildcard Path patterns back a whole family of sibling subcollections from one definition (see below)
• Access control — Per-document and collection-level access rules via Winche.Sentinel; OR semantics with default-deny (any matching rule that grants allows; nothing grants ⇒ denied); reads filtered post-execution; counts and aggregations require a dedicated Aggregate grant (read access does not imply aggregate access)
• PostgreSQL backend — All data stored as typed JSONB; queries compiled to native PostgreSQL SQL with winche_rank/winche_num/winche_text helper functions; IMMUTABLE functions back expression indexes

Secondary Indexes

An IndexDefinition declares a composite expression index over the winche_* ordering family for a collection. IndexDefinition.Path is either an exact collection path or a wildcard pattern with * at document-id positions, letting one definition back a whole family of sibling subcollections:

// pattern index — one definition backs every user's per-entity subcollection
public sealed class SessionHistoryByStartedAt : IndexDefinition
{
    public override string Path => "userData/*/sessionHistory";
    public override IReadOnlyList<IndexField> Fields => [new("startedAt", SortDirection.Desc)];
}

You may pin some ancestor ids and wildcard others (orgs/acme/teams/*/members). Per-entity queries (collection = "userData/alice/sessionHistory") and per-entity aggregation pipelines use the matching pattern index automatically — no change to query code.

Pattern grammar (violations throw InvalidPathPatternException at schema-sync):

• A collection path has an odd number of non-empty segments.
• * is allowed only at document-id positions and must be a whole segment (no a*, **).
• Literal segments match [A-Za-z0-9_-]+.

Cross-collection ("collection group") queries — querying a field across all members of a family at once — are not yet supported.

Access Rules

Access rules determine whether a caller can perform an operation. Implement DocumentAccessRule and register it with AddDocumentAccessRule<T>().

using Winche.Database.Abstraction;
using Winche.Database.Documents;
using Winche.Sentinel.Models;

public class OwnerReadRule : DocumentAccessRule
{
    // Path pattern: literal segments and {param} wildcards; ** matches any depth
    public override string Path => "users/{userId}";

    public override IReadOnlySet<AccessOperation> Operations =>
        new HashSet<AccessOperation> { AccessOperation.Read };

    public override Task<bool> EvaluateAsync(AccessContext<Document> context, CancellationToken ct)
    {
        var uid = context.Claims.TryGetValue("uid", out var v) ? v as string : null;
        context.Params.TryGetValue("userId", out var userId);
        return Task.FromResult(uid != null && uid == userId);
    }
}

Semantics: Rules grant access; there is no explicit deny (Firestore-style). A request is allowed if any rule whose path pattern and Operations set match returns true (OR). A matching rule that returns false does not veto — it simply doesn't grant. If no rule grants, access is denied (default-deny). Registration order does not affect the decision.

Because a grant cannot be revoked by another rule, grant narrowly: don't write a broad ** read grant and expect a more specific rule to restrict it — instead grant read only where it should be allowed (e.g. an owner-scoped rule like the one above) and let default-deny cover the rest.

The operations are Read, Write, Delete, and Aggregate.

Query access is checked per-document after the query runs — documents for which the caller is denied are silently dropped from the result set, so partial results are returned rather than an error.

Aggregation access is gated by the dedicated Aggregate operation, checked at the collection level on the collection of every match and lookup stage before the pipeline runs. It is deny-by-default and independent of Read: granting read access to a collection does not authorize aggregating over it, because an aggregate result (count/sum/min/max, or push/first) can reveal information about documents the caller cannot read individually. Every collection whose rows can reach the output — including lookup targets — needs its own Aggregate grant. Individual result rows are not filtered; the collection-level grant is the entire boundary.

Claims Accessor

To supply per-request caller claims, implement DocumentClaimsAccessor (from Winche.Database.AspNetCore):

using Winche.Database.AspNetCore.Abstraction;

public class CallerClaimsAccessor : DocumentClaimsAccessor
{
    public override IReadOnlyDictionary<string, object?> MapClaims(HttpContext httpContext)
    {
        var uid = httpContext.User.FindFirst("sub")?.Value;
        return uid is null
            ? ImmutableDictionary<string, object?>.Empty
            : new Dictionary<string, object?> { ["uid"] = uid };
    }
}

Register it once and it is shared by both the REST and WebSocket transports:

config.SetCallerClaimsAccessor<CallerClaimsAccessor>();

Document Store Hooks

Hooks let you react to document mutations. Implement DocumentStoreHook and register with AddDocumentStoreHook<T>(). Hooks execute inline and sequentially inside HookFeedConsumer, which is driven by a durable feed runner. Delivery is true at-least-once end-to-end — the cursor advances only after all hooks in a batch succeed, and failed batches are retried with capped backoff (1 s → 30 s). Implementations must be idempotent (use the document version as an idempotency key).

using Winche.Database.Abstraction;
using Winche.Database.Documents;

public class AuditHook : DocumentStoreHook
{
    // Limit this hook to a specific subtree; use "**" to match everything
    public override string Path => "orders/**";

    public override Task OnDocumentSetAsync(string path, Document document, CancellationToken ct)
    {
        // called after a document is created or replaced
        return Task.CompletedTask;
    }

    public override Task OnDocumentUpdatedAsync(string path, Document document, CancellationToken ct)
    {
        // called after a document is patched
        return Task.CompletedTask;
    }

    public override Task OnDocumentDeletedAsync(string path, CancellationToken ct)
    {
        // called after a document (or subtree) is deleted
        return Task.CompletedTask;
    }
}

IDocumentDatabase

The primary service. Inject IDocumentDatabase to interact with the store from application code.

Guarded vs. core. IDocumentDatabase is bound in DI to GuardedDocumentDatabase, which enforces access rules on every operation. The concrete DocumentDatabase is the rule-free core that the guard decorates — inject it only in trusted server-side code (seeding, migrations, internal jobs) that should deliberately bypass authorization. Application code subject to access rules must depend on IDocumentDatabase, never the concrete DocumentDatabase.

public interface IDocumentDatabase
{
    // Reads
    Task<Document?> GetAsync(string path, CancellationToken ct = default);
    Task<IReadOnlyList<Document?>> GetAllAsync(IReadOnlyList<string> paths, CancellationToken ct = default);
    Task<QueryResult> QueryAsync(Query query, CancellationToken ct = default);
    Task<long> CountAsync(Query query, CancellationToken ct = default);
    Task<PipelineResult> AggregateAsync(Pipeline pipeline, CancellationToken ct = default);

    // Writes — every mutation is a Write[]; singles are sugar
    Task<IReadOnlyList<WriteResult>> WriteAsync(IReadOnlyList<Write> writes, CancellationToken ct = default);

    // Transactions (optimistic)
    Task<TransactionHandle> BeginTransactionAsync(CancellationToken ct = default);
    Task<Document?> GetAsync(string transactionId, string path, CancellationToken ct = default);
    Task<QueryResult> QueryAsync(string transactionId, Query query, CancellationToken ct = default);
    Task<IReadOnlyList<WriteResult>> CommitTransactionAsync(string transactionId, IReadOnlyList<Write> writes, CancellationToken ct = default);
    Task RollbackTransactionAsync(string transactionId, CancellationToken ct = default);
    Task<T> RunTransactionAsync<T>(Func<TransactionContext, Task<T>> body, TransactionOptions? options = null, CancellationToken ct = default);

    // Live queries
    IQueryListener Listen(Query query, ListenOptions? options = null);
}

WriteBatch is a fluent helper — construct it with new WriteBatch(db):

await new WriteBatch(db)
    .Set("users/u1", new Dictionary<string, Value> { ["name"] = new StringValue("Alice") })
    .Set("counters/c1", new Dictionary<string, Value> { ["n"] = new IntegerValue(0) })
    .CommitAsync();

Query API

All queries use the typed Query; the wire JSON and C# AST share the same shape. See PROTOCOL for the complete filter/operator reference.

using Winche.Database.Querying.Ast;
using Winche.Database.Documents;
using Winche.Database.Values;

var query = new Query(
    Collection: "users",
    Where: new FieldFilter(FieldPath.Parse("score"), FilterOperator.Gte, new IntegerValue(50)),
    OrderBy: [new Ordering(FieldPath.Parse("score"), SortDirection.Desc)],
    Limit: 25
);

var result = await db.QueryAsync(query);
// result.Documents : IReadOnlyList<Document>
// result.HasMore   : bool

Filter operators

Eq, Ne, Gt, Gte, Lt, Lte, In, NotIn, ArrayContains, ArrayContainsAny, ArrayContainsAll, Contains, StartsWith, EndsWith, Regex

Inequality operators match values of the same type-class only.

Logical operators

And, Or, Not (via CompositeFilter)

Unary operators

IsNull, IsNan, Exists (via UnaryFilter)

Cursor pagination

var cursor = new Cursor(Values: [new IntegerValue(lastScore), new StringValue("users/" + lastId)], Before: false);
var nextPage = new Query("users", OrderBy: [...], Limit: 25, Start: cursor);

Before: false = startAfter (exclusive); Before: true = startAt (inclusive). Mirror for End.

Counting documents

CountAsync runs a COUNT(*) over the same match as QueryAsync, returning a long instead of materializing documents — reusing the Query's collection, filter, and cursor bounds:

var total   = await db.CountAsync(new Query("users"));                                   // whole collection
var active  = await db.CountAsync(new Query("users",
    Where: new FieldFilter(FieldPath.Parse("score"), FilterOperator.Gte, new IntegerValue(50))));
var capped  = await db.CountAsync(new Query("users", Limit: 1000));                       // count is capped at 1000

An explicit Limit caps the count (Firestore count() semantics); an absent limit counts the full match. Like aggregation, counting is gated by the Aggregate access grant (a count can reveal information about documents the caller cannot read individually) and is enforced at the collection level — row-level read rules are not applied, so scope the count by adding the constraining filter to the query itself.

Aggregation Pipelines

using Winche.Database.Querying.Ast;

var pipeline = new Pipeline([
    new Match("orders", Where: null),
    new Lookup("users", FieldPath.Parse("userId"), FieldPath.Parse("id"), "user"),
    new Unwind(FieldPath.Parse("user"), "user"),
    new Group(
        Keys: [new GroupKey("status", FieldPath.Parse("status"))],
        Accumulators: [new Accumulator("total", AggFunction.Sum, FieldPath.Parse("amount"))]
    ),
    new Sort([new Ordering(FieldPath.Parse("total"), SortDirection.Desc)]),
    new Limit(10),
]);

var result = await db.AggregateAsync(pipeline);
// result.Rows : IReadOnlyList<IReadOnlyDictionary<string, Value>>

Available stages: Match, Where, Lookup, Unwind, Group, Project, Sort, Limit, Skip.

Available accumulator functions: Count, Sum, Avg, Min, Max, Push, AddToSet, First, Last.

Transactions

// Automatic retry on ABORTED (up to 5 attempts with backoff)
var result = await db.RunTransactionAsync(async tx =>
{
    var doc = await tx.GetAsync("accounts/a1");
    var balance = ((IntegerValue)doc!.Fields["balance"]).Value;
    tx.Set("accounts/a1", new Dictionary<string, Value> { ["balance"] = new IntegerValue(balance - 100) });
    tx.Set("accounts/a2", ...);
    return balance;
});

// Manual lifecycle
var handle = await db.BeginTransactionAsync();
var doc    = await db.GetAsync(handle.Id, "accounts/a1");  // recorded read
await db.CommitTransactionAsync(handle.Id, [new SetWrite { ... }]);

Transaction configuration (WincheDatabaseOptions.TransactionConfig):

Multi-node note: Transaction state lives in the serving node's in-memory ledger. Multi-node deployments must use sticky routing (all requests in a transaction to the same node) or use the WebSocket API (connection-pinned). Routing violations surface as ABORTED — never corruption.

REST API

Mapped under /documents by default (configurable via MapWincheDatabaseRestApi(prefix: "...")). Document paths in URL parameters are Base64-encoded (UTF-8 bytes).

Convenience endpoints:

Colon-verb endpoints:

Access rules are enforced on all routes. The claims accessor runs as an endpoint filter before every request.

See PROTOCOL for full request/response examples.

WebSocket API

Connect at /documents/ws. All operations are exchanged as typed JSON messages over a single WebSocket. Handshake: send {"type":"hello","protocol":3} — receive {"type":"welcome","connectionId":"...","protocol":3}.

See PROTOCOL for full message shapes, close codes, and listener protocol details.

Requirements

• .NET 10.0
• PostgreSQL (any recent version with JSONB support)

License

Elastic License 2.0