< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Channels.MongoDB.MongoChannelSubscriberDocument
Assembly: AsyncResponse.Channels.MongoDB
File(s): /_/src/Channels/AsyncResponse.Channels.MongoDB/MongoDbChannelStore.cs
Line coverage
0%
Covered lines: 0
Uncovered lines: 5
Coverable lines: 5
Total lines: 915
Line coverage: 0%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_Id()100%210%
get_CorrelationId()100%210%
get_RegistrationId()100%210%
get_InstanceId()100%210%
get_ExpiresAtUtc()100%210%

File(s)

/_/src/Channels/AsyncResponse.Channels.MongoDB/MongoDbChannelStore.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using Microsoft.Extensions.Logging.Abstractions;
 3using Microsoft.Extensions.Options;
 4using MongoDB.Bson;
 5using MongoDB.Bson.Serialization.Attributes;
 6using MongoDB.Driver;
 7using System.Runtime.CompilerServices;
 8using AsyncResponse.Internal;
 9
 10namespace AsyncResponse.Channels.MongoDB;
 11
 12/// <summary>One stored response envelope row/document as the channel store returns it.</summary>
 13/// <remarks>
 14/// <c>EnvelopeJson</c> is the stored envelope, or <c>null</c> for a document the dispatch sweep loaded header-only (an
 15/// already-acknowledged one — see <see cref="MongoDbChannelStore.LoadMessagesAsync"/>); the
 16/// sweep hydrates the few such documents it still has to deliver through
 17/// <see cref="MongoDbChannelStore.LoadMessagesByIdAsync"/> before handing them to a waiter.
 18/// </remarks>
 19internal readonly record struct MongoDbChannelMessage(
 20    Guid Id,
 21    string CorrelationId,
 22    string? EnvelopeJson,
 23    DateTimeOffset CreatedAtUtc,
 24    DateTimeOffset? AckedAtUtc = null,
 25    long? AckedSeq = null);
 26
 27/// <summary>Document adapter for the MongoDB channel collections and change-stream wake.</summary>
 28internal sealed class MongoDbChannelStore : IDisposable
 29{
 30    private readonly IMongoCollection<MongoRecoveryStateDocument> _recovery;
 31    private readonly IMongoCollection<MongoChannelMessageDocument> _messages;
 32    private readonly IMongoCollection<MongoChannelSubscriberDocument> _subscribers;
 33    private readonly IMongoCollection<BsonDocument> _counters;
 34    private readonly IMongoDatabase _database;
 35    private readonly MongoDbAsyncResponseChannelOptions _options;
 36    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 37    private readonly IMongoClient? _ownedClient;
 38    private readonly ILogger _logger;
 39    private bool _created;
 40
 41    public MongoDbChannelStore(
 42        IMongoDatabase database,
 43        IOptions<MongoDbAsyncResponseChannelOptions> options,
 44        IMongoClient? ownedClient = null,
 45        IMongoNamespaceRegistry? namespaceRegistry = null,
 46        ILogger? logger = null)
 47    {
 48        _options = options.Value;
 49        _options.Validate();
 50        _database = database;
 51        _logger = logger ?? NullLogger.Instance;
 52
 53        // Cross-component collection ownership (DI-hosted stores only): another AsyncResponse
 54        // component configured onto one of these collections — the derived counters collection
 55        // included — must fail startup in either construction order.
 56        namespaceRegistry?.Claim(
 57            MongoNamespaceRegistry.ClusterKey(database),
 58            database.DatabaseNamespace.DatabaseName,
 59            "MongoDB channel",
 60            [
 61                (_options.RecoveryStateCollection, nameof(_options.RecoveryStateCollection)),
 62                (_options.MessageCollection, nameof(_options.MessageCollection)),
 63                (_options.SubscriberCollection, nameof(_options.SubscriberCollection)),
 64                (CountersCollectionName(_options.MessageCollection), "derived ack-counter collection"),
 65            ]);
 66
 67        // Namespace BYTE limits can only be checked here, where the actual database name is
 68        // first known — and the derived counters namespace is 9 bytes longer than the configured
 69        // message collection, so a near-limit configuration passed every static check and failed
 70        // at the first ack-sequence draw.
 71        MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.RecoveryStateCollection, nameof(_options.Re
 72        MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.MessageCollection, nameof(_options.MessageC
 73        MongoNamespaceRegistry.ValidateEffectiveNamespace(database, _options.SubscriberCollection, nameof(_options.Subsc
 74        MongoNamespaceRegistry.ValidateEffectiveNamespace(database, CountersCollectionName(_options.MessageCollection), 
 75
 76        // Primary reads override whatever the host-registered client/connection string configured:
 77        // a secondaryPreferred connection routed the liveness probe, recovery-state read and message
 78        // load to a lagging secondary, so a publisher racing a fresh registration saw 0 subscribers
 79        // AND 0 recovery states and dropped the response while reporting success. It also keeps
 80        // reads on the same authority whose $$NOW the expiry filters evaluate against (same
 81        // reasoning as the MongoDB durable-flow store's pin).
 82        _recovery = database.GetCollection<MongoRecoveryStateDocument>(_options.RecoveryStateCollection)
 83            .WithReadPreference(ReadPreference.Primary);
 84        _messages = database.GetCollection<MongoChannelMessageDocument>(_options.MessageCollection)
 85            .WithReadPreference(ReadPreference.Primary);
 86        _subscribers = database.GetCollection<MongoChannelSubscriberDocument>(_options.SubscriberCollection)
 87            .WithReadPreference(ReadPreference.Primary);
 88        // The monotonic ack sequence: delivery claims and subscription registrations draw from
 89        // this ONE counter, giving acked_seq and a subscription's start position a total order no
 90        // pair of same-tick timestamps has. Created on first upsert; no index needed (_id only).
 91        _counters = database.GetCollection<BsonDocument>(CountersCollectionName(_options.MessageCollection))
 92            .WithReadPreference(ReadPreference.Primary);
 93        _ownedClient = ownedClient;
 94    }
 95
 96    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 97    {
 98        if (_created)
 99            return;
 100
 101        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 102        try
 103        {
 104            if (_created)
 105                return;
 106
 107            // Persisted cross-host ownership: the in-container registry cannot see other hosts
 108            // or directly constructed stores, so claim the effective collections here — before
 109            // any index DDL, and INDEPENDENTLY of AutoCreateIndexes (disabling index DDL must
 110            // not disable collision protection) — and fail startup when another component
 111            // already owns one.
 112            if (_options.UseOwnershipLedger)
 113            {
 114                await MongoOwnershipLedger.ClaimAsync(
 115                    _database,
 116                    "MongoDB channel",
 117                    [
 118                        (_options.RecoveryStateCollection, nameof(_options.RecoveryStateCollection)),
 119                        (_options.MessageCollection, nameof(_options.MessageCollection)),
 120                        (_options.SubscriberCollection, nameof(_options.SubscriberCollection)),
 121                        (CountersCollectionName(_options.MessageCollection), "derived ack-counter collection"),
 122                    ],
 123                    cancellationToken).ConfigureAwait(false);
 124            }
 125
 126            if (!_options.AutoCreateIndexes)
 127            {
 128                // Manually managed indexes get a one-time read-only check instead of DDL. There
 129                // is no collection shape to verify (documents are schemaless), so the silent
 130                // failure modes are all indexes — above all a missing TTL index, which means
 131                // nothing ever reaps expired documents. Absence is a warning, never a startup
 132                // failure: indexes degrade retention and performance, not correctness, and a
 133                // least-privilege operator may provision them out of band.
 134                await WarnIfManagedIndexesMissingAsync(cancellationToken).ConfigureAwait(false);
 135                _created = true;
 136                return;
 137            }
 138
 139            // TTL indexes (expireAfterSeconds = 0 on the expiry timestamp) make MongoDB itself reap
 140            // expired documents — no application-side pruning needed. Reads still filter on the
 141            // expiry because the TTL monitor only runs periodically (~60s).
 142            await CreateTtlIndexAsync(
 143                _recovery,
 144                Builders<MongoRecoveryStateDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc),
 145                $"{_options.RecoveryStateCollection}_expires_idx",
 146                cancellationToken).ConfigureAwait(false);
 147            await _recovery.Indexes.CreateOneAsync(
 148                new CreateIndexModel<MongoRecoveryStateDocument>(
 149                    Builders<MongoRecoveryStateDocument>.IndexKeys
 150                        .Ascending(item => item.CorrelationId)
 151                        .Ascending(item => item.RegisteredAtUtc),
 152                    new CreateIndexOptions { Name = $"{_options.RecoveryStateCollection}_correlation_idx" }),
 153                cancellationToken: cancellationToken).ConfigureAwait(false);
 154
 155            await CreateTtlIndexAsync(
 156                _messages,
 157                Builders<MongoChannelMessageDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc),
 158                $"{_options.MessageCollection}_expires_idx",
 159                cancellationToken).ConfigureAwait(false);
 160            await _messages.Indexes.CreateOneAsync(
 161                new CreateIndexModel<MongoChannelMessageDocument>(
 162                    Builders<MongoChannelMessageDocument>.IndexKeys
 163                        .Ascending(item => item.CorrelationId)
 164                        .Ascending(item => item.CreatedAtUtc),
 165                    new CreateIndexOptions { Name = $"{_options.MessageCollection}_correlation_created_idx" }),
 166                cancellationToken: cancellationToken).ConfigureAwait(false);
 167
 168            await CreateTtlIndexAsync(
 169                _subscribers,
 170                Builders<MongoChannelSubscriberDocument>.IndexKeys.Ascending(item => item.ExpiresAtUtc),
 171                $"{_options.SubscriberCollection}_expires_idx",
 172                cancellationToken).ConfigureAwait(false);
 173            await _subscribers.Indexes.CreateOneAsync(
 174                new CreateIndexModel<MongoChannelSubscriberDocument>(
 175                    Builders<MongoChannelSubscriberDocument>.IndexKeys.Ascending(item => item.CorrelationId),
 176                    new CreateIndexOptions { Name = $"{_options.SubscriberCollection}_correlation_idx" }),
 177                cancellationToken: cancellationToken).ConfigureAwait(false);
 178
 179            _created = true;
 180        }
 181        finally
 182        {
 183            _ensureGate.Release();
 184        }
 185    }
 186
 187    private static async Task CreateTtlIndexAsync<TDocument>(
 188        IMongoCollection<TDocument> collection,
 189        IndexKeysDefinition<TDocument> keys,
 190        string indexName,
 191        CancellationToken cancellationToken)
 192    {
 193        var model = new CreateIndexModel<TDocument>(
 194            keys,
 195            new CreateIndexOptions { Name = indexName, ExpireAfter = TimeSpan.Zero });
 196        try
 197        {
 198            await collection.Indexes.CreateOneAsync(model, cancellationToken: cancellationToken).ConfigureAwait(false);
 199        }
 200        catch (MongoCommandException ex) when (ex.Code is 85 or 86)
 201        {
 202            // IndexOptionsConflict/IndexKeySpecsConflict: an earlier deployment created the
 203            // same-named index with different options. Replace it in place.
 204            try
 205            {
 206                await collection.Indexes.DropOneAsync(indexName, cancellationToken).ConfigureAwait(false);
 207            }
 208            catch (MongoCommandException dropException) when (dropException.Code == 27)
 209            {
 210                // IndexNotFound: a peer host in the same rolling deploy took the same branch and
 211                // dropped it first. Converge on the recreate below (idempotent for an identical spec).
 212            }
 213
 214            await collection.Indexes.CreateOneAsync(model, cancellationToken: cancellationToken).ConfigureAwait(false);
 215        }
 216    }
 217
 218    private async Task WarnIfManagedIndexesMissingAsync(CancellationToken cancellationToken)
 219    {
 220        try
 221        {
 222            await WarnIfCollectionIndexesMissingAsync(_recovery, _options.RecoveryStateCollection, cancellationToken).Co
 223            await WarnIfCollectionIndexesMissingAsync(_messages, _options.MessageCollection, cancellationToken).Configur
 224            await WarnIfCollectionIndexesMissingAsync(_subscribers, _options.SubscriberCollection, cancellationToken).Co
 225        }
 226        catch (Exception ex) when (ex is not OperationCanceledException)
 227        {
 228            // A deployment that cannot even list indexes (no listIndexes privilege, server
 229            // unreachable at first use) must not lose the actual operation to the check: the
 230            // caller's own store call surfaces any real connectivity failure.
 231            _logger.LogDebug(ex, "Skipping index verification for the manually managed MongoDB channel collections; list
 232        }
 233    }
 234
 235    private async Task WarnIfCollectionIndexesMissingAsync<TDocument>(
 236        IMongoCollection<TDocument> collection,
 237        string collectionName,
 238        CancellationToken cancellationToken)
 239    {
 240        List<BsonDocument> indexes;
 241        try
 242        {
 243            using var cursor = await collection.Indexes.ListAsync(cancellationToken).ConfigureAwait(false);
 244            indexes = await cursor.ToListAsync(cancellationToken).ConfigureAwait(false);
 245        }
 246        catch (MongoCommandException ex) when (ex.Code == 26)
 247        {
 248            // NamespaceNotFound: the collection does not exist yet. MongoDB creates it bare on
 249            // the first write — which, with index DDL disabled, is exactly a collection with no
 250            // TTL index.
 251            indexes = [];
 252        }
 253
 254        // Matched by KEY, not by name: operators own the naming of manually provisioned indexes.
 255        if (!indexes.Any(index => IndexLeadsOn(index, "expires_at") && index.Contains("expireAfterSeconds")))
 256        {
 257            _logger.LogWarning(
 258                "MongoDB collection {Database}.{Collection} has no TTL index on 'expires_at' and AutoCreateIndexes is di
 259                "Expired documents are never reaped, so the collection grows without bound. " +
 260                "Create a TTL index (expireAfterSeconds: 0 on 'expires_at') or enable AutoCreateIndexes.",
 261                _database.DatabaseNamespace.DatabaseName, collectionName);
 262        }
 263
 264        if (!indexes.Any(index => IndexLeadsOn(index, "correlation_id")))
 265        {
 266            _logger.LogWarning(
 267                "MongoDB collection {Database}.{Collection} has no index leading on 'correlation_id' and AutoCreateIndex
 268                "Correlation-id lookups scan the whole collection — performance only; create the index to restore indexe
 269                _database.DatabaseNamespace.DatabaseName, collectionName);
 270        }
 271    }
 272
 273    /// <summary>
 274    /// Whether the listed index's FIRST key field is <paramref name="field"/> — what a prefix
 275    /// lookup uses, and (TTL indexes being single-field) what identifies the TTL index.
 276    /// </summary>
 277    private static bool IndexLeadsOn(BsonDocument index, string field)
 278        => index.TryGetValue("key", out var key)
 279           && key is BsonDocument keyDocument
 280           && keyDocument.ElementCount > 0
 281           && keyDocument.GetElement(0).Name == field;
 282
 283    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 284    {
 285        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 286        // Upsert pipeline stamped with the server clock ($$NOW), matching the message-side
 287        // discipline (and the PG/SqlServer DB-clock discipline): app-clock expiry math would shift
 288        // the recovery window by whatever the client clock is skewed.
 289        await _recovery.UpdateOneAsync(
 290            Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, state.Registr
 291            BuildRecoveryStateUpsertPipeline(correlationId, state, ttl),
 292            new UpdateOptions { IsUpsert = true },
 293            cancellationToken).ConfigureAwait(false);
 294    }
 295
 296    /// <summary>
 297    /// Upsert pipeline for a recovery registration: every field is overwritten (a re-save refreshes
 298    /// the registration), with <c>expires_at</c>/<c>registered_at</c> computed on the server clock.
 299    /// </summary>
 300    internal static UpdateDefinition<MongoRecoveryStateDocument> BuildRecoveryStateUpsertPipeline(
 301        string correlationId,
 302        RecoveryState state,
 303        TimeSpan ttl)
 304        => Builders<MongoRecoveryStateDocument>.Update.Pipeline(new[]
 305        {
 306            new BsonDocument("$set", new BsonDocument
 307            {
 308                ["correlation_id"] = correlationId,
 309                ["registration_id"] = new BsonBinaryData(state.RegistrationId, GuidRepresentation.Standard),
 310                ["state_json"] = AsyncResponseJson.Serialize(state),
 311                ["expires_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", ttl.TotalMilliseconds }),
 312                ["registered_at"] = "$$NOW"
 313            })
 314        });
 315
 316    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 317    {
 318        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 319        var filter = Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.CorrelationId, correlationId)
 320                     & NotExpiredOnServerClock<MongoRecoveryStateDocument>();
 321        var documents = await _recovery.Find(filter)
 322            .SortBy(item => item.RegisteredAtUtc)
 323            .Project(item => item.StateJson)
 324            .ToListAsync(cancellationToken).ConfigureAwait(false);
 325        return documents;
 326    }
 327
 328    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 329    {
 330        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 331        var single = await _recovery.DeleteOneAsync(
 332            Builders<MongoRecoveryStateDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrationI
 333            cancellationToken).ConfigureAwait(false);
 334        return single.DeletedCount > 0;
 335    }
 336
 337    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([EnumeratorCancellation] CancellationToken cancella
 338    {
 339        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 340        var filter = NotExpiredOnServerClock<MongoRecoveryStateDocument>();
 341        using var cursor = await _recovery.Find(filter)
 342            .SortBy(item => item.RegisteredAtUtc)
 343            .Project(item => item.StateJson)
 344            .ToCursorAsync(cancellationToken).ConfigureAwait(false);
 345        while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
 346        {
 347            foreach (var json in cursor.Current)
 348                yield return json;
 349        }
 350    }
 351
 352    /// <summary>
 353    /// Inserts a response envelope document. The caller supplies the message id and the write is an
 354    /// upsert that preserves an existing document, so a retried publish is idempotent rather than
 355    /// duplicating the response. Timestamps are stamped with the server clock (<c>$$NOW</c>) so
 356    /// dispatch watermarks never mix client and server clocks. The insert itself is the wake signal:
 357    /// every process's change stream observes it.
 358    /// </summary>
 359    public Task<MongoDbChannelMessage> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpan r
 360        => AsyncResponseRetry.ExecuteAsync(
 361            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 362            MongoTransientFaults.IsTransient,
 363            _options.PublishMaxAttempts,
 364            _options.PublishRetryBaseDelay,
 365            _options.PublishRetryMaxDelay,
 366            cancellationToken);
 367
 368    private async Task<MongoDbChannelMessage> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJson,
 369    {
 370        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 371
 372        // findOneAndUpdate instead of updateOne so the returned document carries the
 373        // server-stamped ($$NOW) created_at — the original document's on a publish retry, per the
 374        // pipeline's $ifNull, together with its settlement columns — for the same-process fast
 375        // path's watermark comparison (a fabricated null acked_at replayed an already-consumed
 376        // response to a waiter registered after the ack).
 377        var document = await _messages.FindOneAndUpdateAsync(
 378            Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, id),
 379            BuildInsertMessagePipeline(correlationId, envelopeJson, retention),
 380            new FindOneAndUpdateOptions<MongoChannelMessageDocument>
 381            {
 382                IsUpsert = true,
 383                ReturnDocument = ReturnDocument.After
 384            },
 385            cancellationToken).ConfigureAwait(false);
 386
 387        // Upsert + ReturnDocument.After cannot return null from a healthy server. If a driver
 388        // anomaly ever surfaces one, persistence is UNKNOWN — reporting success with a fabricated
 389        // app-clock timestamp would both lie about it and feed a client clock into the
 390        // server-clock watermark. Fail instead, so the retry/error path runs.
 391        return document is null
 392            ? throw new InvalidOperationException(
 393                $"MongoDB response upsert for message {id} returned no document despite IsUpsert + ReturnDocument.After;
 394            : new MongoDbChannelMessage(
 395                id,
 396                correlationId,
 397                envelopeJson,
 398                new DateTimeOffset(document.CreatedAtUtc, TimeSpan.Zero),
 399                document.AckedAtUtc is { } acked ? new DateTimeOffset(acked, TimeSpan.Zero) : null,
 400                document.AckedSeq);
 401    }
 402
 403    /// <summary>
 404    /// Upsert pipeline for a response envelope: <c>$ifNull</c> keeps the original server-stamped
 405    /// timestamps and claim flags when a publish retry finds the document already present.
 406    /// </summary>
 407    internal static UpdateDefinition<MongoChannelMessageDocument> BuildInsertMessagePipeline(
 408        string correlationId,
 409        string envelopeJson,
 410        TimeSpan retention)
 411        => Builders<MongoChannelMessageDocument>.Update.Pipeline(new[]
 412        {
 413            new BsonDocument("$set", new BsonDocument
 414            {
 415                ["correlation_id"] = correlationId,
 416                ["envelope_json"] = envelopeJson,
 417                ["created_at"] = new BsonDocument("$ifNull", new BsonArray { "$created_at", "$$NOW" }),
 418                ["expires_at"] = new BsonDocument("$ifNull", new BsonArray
 419                {
 420                    "$expires_at",
 421                    new BsonDocument("$add", new BsonArray { "$$NOW", retention.TotalMilliseconds })
 422                }),
 423                ["acked_at"] = new BsonDocument("$ifNull", new BsonArray { "$acked_at", BsonNull.Value }),
 424                ["recovery_claimed"] = new BsonDocument("$ifNull", new BsonArray { "$recovery_claimed", false })
 425            })
 426        });
 427
 428    public async Task<IReadOnlyList<MongoDbChannelMessage>> LoadMessagesAsync(
 429        string correlationId,
 430        DateTimeOffset sinceUtc,
 431        int batchSize,
 432        DateTimeOffset? afterCreatedAtUtc,
 433        Guid? afterId,
 434        CancellationToken cancellationToken)
 435    {
 436        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 437        var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CorrelationId, correlationId)
 438                     & Builders<MongoChannelMessageDocument>.Filter.Gte(item => item.CreatedAtUtc, sinceUtc.UtcDateTime)
 439                     & NotExpiredOnServerClock<MongoChannelMessageDocument>();
 440        if (afterCreatedAtUtc is not null)
 441        {
 442            var afterCreated = afterCreatedAtUtc.Value.UtcDateTime;
 443            var cursorId = afterId ?? throw new ArgumentNullException(nameof(afterId));
 444            filter &= Builders<MongoChannelMessageDocument>.Filter.Or(
 445                Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.CreatedAtUtc, afterCreated),
 446                Builders<MongoChannelMessageDocument>.Filter.And(
 447                    Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CreatedAtUtc, afterCreated),
 448                    Builders<MongoChannelMessageDocument>.Filter.Gt(item => item.Id, cursorId)));
 449        }
 450        var documents = await _messages.Find(filter)
 451            .Project(SweepProjection)
 452            .Sort(Builders<MongoChannelMessageDocument>.Sort
 453                .Ascending(item => item.CreatedAtUtc)
 454                .Ascending(item => item.Id))
 455            .Limit(batchSize)
 456            .ToListAsync(cancellationToken).ConfigureAwait(false);
 457        return ToMessages(documents);
 458    }
 459
 460    /// <summary>
 461    /// The sweep's projection: every field but the envelope, and the envelope only for a document
 462    /// nobody has acknowledged yet. Acknowledged documents are the consumed history the sweep
 463    /// re-reads on every tick (they stay in the result so a fan-out waiter in ANOTHER process
 464    /// still receives them): shipping their bodies with each sweep made a long-lived progress
 465    /// subscription's cost grow with its whole retained history. The shared sweep fetches the
 466    /// envelope by id for the rare acknowledged document a live subscription has not seen.
 467    /// <c>$ifNull</c> folds a missing <c>acked_at</c> (a pre-settlement document) into null.
 468    /// </summary>
 469    internal static readonly ProjectionDefinition<MongoChannelMessageDocument, MongoChannelMessageDocument> SweepProject
 470        new BsonDocumentProjectionDefinition<MongoChannelMessageDocument, MongoChannelMessageDocument>(new BsonDocument
 471        {
 472            ["_id"] = 1,
 473            ["correlation_id"] = 1,
 474            ["created_at"] = 1,
 475            ["expires_at"] = 1,
 476            ["acked_at"] = 1,
 477            ["acked_seq"] = 1,
 478            ["recovery_claimed"] = 1,
 479            ["envelope_json"] = new BsonDocument("$cond", new BsonArray
 480            {
 481                new BsonDocument("$eq", new BsonArray
 482                {
 483                    new BsonDocument("$ifNull", new BsonArray { "$acked_at", BsonNull.Value }),
 484                    BsonNull.Value
 485                }),
 486                "$envelope_json",
 487                BsonNull.Value
 488            })
 489        });
 490
 491    /// <summary>
 492    /// The full documents (envelope included) for <paramref name="ids"/> under
 493    /// <paramref name="correlationId"/>, in sweep order — how the dispatch sweep hydrates the
 494    /// header-only acknowledged documents it still has to deliver. A document reaped between the
 495    /// sweep's page and this read is simply absent.
 496    /// </summary>
 497    public async Task<IReadOnlyList<MongoDbChannelMessage>> LoadMessagesByIdAsync(
 498        string correlationId,
 499        IReadOnlyList<Guid> ids,
 500        CancellationToken cancellationToken)
 501    {
 502        if (ids.Count == 0)
 503            return [];
 504
 505        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 506        var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.CorrelationId, correlationId)
 507                     & Builders<MongoChannelMessageDocument>.Filter.In(item => item.Id, ids)
 508                     & NotExpiredOnServerClock<MongoChannelMessageDocument>();
 509        var documents = await _messages.Find(filter)
 510            .Sort(Builders<MongoChannelMessageDocument>.Sort
 511                .Ascending(item => item.CreatedAtUtc)
 512                .Ascending(item => item.Id))
 513            .ToListAsync(cancellationToken).ConfigureAwait(false);
 514        return ToMessages(documents);
 515    }
 516
 517    private static List<MongoDbChannelMessage> ToMessages(List<MongoChannelMessageDocument> documents)
 518    {
 519        var messages = new List<MongoDbChannelMessage>(documents.Count);
 520        foreach (var document in documents)
 521            messages.Add(new MongoDbChannelMessage(
 522                document.Id,
 523                document.CorrelationId,
 524                document.EnvelopeJson,
 525                new DateTimeOffset(document.CreatedAtUtc, TimeSpan.Zero),
 526                document.AckedAtUtc is { } acked ? new DateTimeOffset(acked, TimeSpan.Zero) : null,
 527                document.AckedSeq));
 528        return messages;
 529    }
 530
 531    /// <summary>
 532    /// Atomically claims a message for live delivery via <c>findOneAndUpdate</c>: sets
 533    /// <c>acked_at</c> unless the publisher has already routed it to the lost-subscriber path
 534    /// (<c>recovery_claimed</c>). Returns <c>false</c> when recovery owns the message, so a
 535    /// slow-but-live waiter does not deliver a response the recovery callback already handled.
 536    /// Multiple processes may each win this claim, preserving cross-process fan-out, because it
 537    /// gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>.
 538    /// </summary>
 539    public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken)
 540    {
 541        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 542        // The sequence value is drawn BEFORE the claim lands, so its position reflects when this
 543        // delivery happened relative to subscription registrations (which draw from the same
 544        // counter). An unused draw on an already-acked row leaves a harmless gap; $ifNull keeps
 545        // the FIRST claim's stamp, mirroring acked_at.
 546        var ackSeq = await DrawAckSequenceAsync(cancellationToken).ConfigureAwait(false);
 547        var claimed = await _messages.FindOneAndUpdateAsync(
 548            BuildDeliveryClaimFilter(messageId),
 549            BuildDeliveryClaimUpdate(ackSeq),
 550            new FindOneAndUpdateOptions<MongoChannelMessageDocument> { ReturnDocument = ReturnDocument.After },
 551            cancellationToken).ConfigureAwait(false);
 552        return claimed is not null;
 553    }
 554
 555    internal static FilterDefinition<MongoChannelMessageDocument> BuildDeliveryClaimFilter(Guid messageId)
 556        => Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId)
 557           & Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.RecoveryClaimed, false)
 558           & NotExpiredOnServerClock<MongoChannelMessageDocument>();
 559
 560    internal static UpdateDefinition<MongoChannelMessageDocument> BuildDeliveryClaimUpdate(long ackSeq)
 561        => Builders<MongoChannelMessageDocument>.Update.Pipeline(new[]
 562        {
 563            // Both fields are computed from the PRE-update document (one $set stage), and the
 564            // sequence is stamped ONLY when this same update transitions acked_at from null: a
 565            // row acked by a pre-sequence build must stay permanently unsequenced — back-filling
 566            // it on a later fan-out re-claim would pair an OLD acked_at with a FRESH sequence
 567            // value, and a waiter that registered in the original ack's tick would then read the
 568            // tie as post-registration fan-out, replaying a response its predecessor consumed.
 569            new BsonDocument("$set", new BsonDocument
 570            {
 571                ["acked_at"] = new BsonDocument("$ifNull", new BsonArray { "$acked_at", "$$NOW" }),
 572                ["acked_seq"] = new BsonDocument("$cond", new BsonArray
 573                {
 574                    new BsonDocument("$eq", new BsonArray { new BsonDocument("$ifNull", new BsonArray { "$acked_at", Bso
 575                    ackSeq,
 576                    "$acked_seq"
 577                })
 578            })
 579        });
 580
 581    private async Task<long> DrawAckSequenceAsync(CancellationToken cancellationToken)
 582    {
 583        var counter = await _counters.FindOneAndUpdateAsync<BsonDocument>(
 584            new BsonDocument("_id", "ack_seq"),
 585            new BsonDocument("$inc", new BsonDocument("seq", 1L)),
 586            new FindOneAndUpdateOptions<BsonDocument, BsonDocument> { IsUpsert = true, ReturnDocument = ReturnDocument.A
 587            cancellationToken).ConfigureAwait(false);
 588        return counter["seq"].ToInt64();
 589    }
 590
 591    /// <summary>
 592    /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only
 593    /// while no waiter has delivered (<c>acked_at</c> is still null). Returns <c>true</c> when
 594    /// recovery wins; <c>false</c> means a live waiter already took the message, so the publisher
 595    /// must not also fire the recovery callback. Document-level atomicity of
 596    /// <c>findOneAndUpdate</c> serializes this against <see cref="TryClaimForDeliveryAsync"/>.
 597    /// </summary>
 598    public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken)
 599    {
 600        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 601        var claimed = await _messages.FindOneAndUpdateAsync(
 602            BuildRecoveryClaimFilter(messageId),
 603            Builders<MongoChannelMessageDocument>.Update.Set(item => item.RecoveryClaimed, true),
 604            new FindOneAndUpdateOptions<MongoChannelMessageDocument> { ReturnDocument = ReturnDocument.After },
 605            cancellationToken).ConfigureAwait(false);
 606        return claimed is not null;
 607    }
 608
 609    internal static FilterDefinition<MongoChannelMessageDocument> BuildRecoveryClaimFilter(Guid messageId)
 610        => Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId)
 611           & Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.AckedAtUtc, null);
 612
 613    /// <summary>Returns the server's current UTC time, used as a clock-safe delivery watermark.</summary>
 614    public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken)
 615    {
 616        // Primary, explicitly: the watermark must come from the clock whose $$NOW stamps the rows
 617        // it bounds — a secondary's localTime can lag or skew from the primary's.
 618        var reply = await _database.RunCommandAsync<BsonDocument>(
 619            new BsonDocument("hello", 1),
 620            ReadPreference.Primary,
 621            cancellationToken).ConfigureAwait(false);
 622        return reply.TryGetValue("localTime", out var localTime) && localTime is BsonDateTime serverTime
 623            ? new DateTimeOffset(serverTime.ToUniversalTime(), TimeSpan.Zero)
 624            : DateTimeOffset.UtcNow;
 625    }
 626
 627    /// <summary>
 628    /// A subscription's registration watermark: the server's UTC clock (for the created-at bound)
 629    /// and a fresh position in the monotonic ack sequence (for the exact acked-history bound —
 630    /// see the watermark in the shared channel base). Drawn by ONE atomic counter update whose
 631    /// pipeline advances the sequence and stamps <c>$$NOW</c> in the same document write
 632    /// (PG/SqlServer single-statement parity): with separate clock and sequence round trips, a
 633    /// delivery claim landing between them pairs a same-millisecond <c>acked_at</c> with a lower
 634    /// sequence, and the same-tick tie-breaker then files a legitimate fan-out delivery as
 635    /// history.
 636    /// </summary>
 637    public async Task<(DateTimeOffset ServerTimeUtc, long StartSeq)> GetSubscriptionStartAsync(CancellationToken cancell
 638    {
 639        var counter = await _counters.FindOneAndUpdateAsync<BsonDocument>(
 640            new BsonDocument("_id", "ack_seq"),
 641            BuildSubscriptionStartPipeline(),
 642            new FindOneAndUpdateOptions<BsonDocument, BsonDocument> { IsUpsert = true, ReturnDocument = ReturnDocument.A
 643            cancellationToken).ConfigureAwait(false);
 644
 645        // The pipeline stamps drawn_at from $$NOW unconditionally, so anything else is a driver
 646        // anomaly. Failing beats an app-clock fallback, which would silently feed a client clock
 647        // into the server-clock watermark this draw exists to protect.
 648        return counter is not null
 649               && counter.TryGetValue("drawn_at", out var drawnAt)
 650               && drawnAt is BsonDateTime serverTime
 651            ? (new DateTimeOffset(serverTime.ToUniversalTime(), TimeSpan.Zero), counter["seq"].ToInt64())
 652            : throw new InvalidOperationException(
 653                "MongoDB subscription-start draw returned no server-stamped counter document despite IsUpsert + ReturnDo
 654    }
 655
 656    /// <summary>
 657    /// Counter-update pipeline for a subscription registration: advances the monotonic ack
 658    /// sequence AND stamps the draw with the server clock in one atomic document update, so no
 659    /// delivery claim can interleave between the sequence draw and the clock read.
 660    /// <c>$ifNull</c> seeds a fresh counter document at 1, matching the delivery claim's
 661    /// <c>$inc</c> upsert.
 662    /// </summary>
 663    internal static UpdateDefinition<BsonDocument> BuildSubscriptionStartPipeline()
 664        => Builders<BsonDocument>.Update.Pipeline(new[]
 665        {
 666            new BsonDocument("$set", new BsonDocument
 667            {
 668                ["seq"] = new BsonDocument("$add", new BsonArray { new BsonDocument("$ifNull", new BsonArray { "$seq", 0
 669                ["drawn_at"] = "$$NOW"
 670            })
 671        });
 672
 673    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 674    {
 675        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 676        var filter = Builders<MongoChannelMessageDocument>.Filter.Eq(item => item.Id, messageId)
 677                     & NotExpiredOnServerClock<MongoChannelMessageDocument>();
 678        // Direct member projection, not an anonymous type: anonymous projections lower to the
 679        // RequiresUnreferencedCode Expression.New(ctor, args, members) overload, which the ILC
 680        // trim analysis rejects (Roslyn's analyzer skips compiler-lowered expression trees, so
 681        // only Native AOT publishes catch it). Missing document and unacked document both come
 682        // back as null, which is exactly the contract here.
 683        var ackedAtUtc = await _messages.Find(filter)
 684            .Project(item => item.AckedAtUtc)
 685            .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false);
 686        return ackedAtUtc is not null;
 687    }
 688
 689    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 690    {
 691        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 692        await _subscribers.UpdateOneAsync(
 693            Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrat
 694            BuildSubscriberUpsertPipeline(correlationId, registrationId, instanceId, ttl),
 695            new UpdateOptions { IsUpsert = true },
 696            cancellationToken).ConfigureAwait(false);
 697    }
 698
 699    /// <summary>
 700    /// Upsert pipeline for a subscriber liveness document, with <c>expires_at</c> computed on the
 701    /// server clock ($$NOW) — app-clock liveness math would let a skewed client look dead (or
 702    /// immortal) to publishers comparing against server-side expiry.
 703    /// </summary>
 704    internal static UpdateDefinition<MongoChannelSubscriberDocument> BuildSubscriberUpsertPipeline(
 705        string correlationId,
 706        Guid registrationId,
 707        string instanceId,
 708        TimeSpan ttl)
 709        => Builders<MongoChannelSubscriberDocument>.Update.Pipeline(new[]
 710        {
 711            new BsonDocument("$set", new BsonDocument
 712            {
 713                ["correlation_id"] = correlationId,
 714                ["registration_id"] = new BsonBinaryData(registrationId, GuidRepresentation.Standard),
 715                ["instance_id"] = instanceId,
 716                ["expires_at"] = new BsonDocument("$add", new BsonArray { "$$NOW", ttl.TotalMilliseconds })
 717            })
 718        });
 719
 720    public async Task HeartbeatSubscribersAsync(
 721        string instanceId,
 722        IReadOnlyCollection<(string CorrelationId, Guid RegistrationId)> registrations,
 723        TimeSpan ttl,
 724        CancellationToken cancellationToken)
 725    {
 726        if (registrations.Count == 0)
 727            return;
 728
 729        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 730
 731        // Per-registration upserts rather than one bare UpdateMany: the caller only heartbeats
 732        // registrations that are live in this process, so a missing document means the TTL reaper
 733        // deleted it (e.g. after a >timeout stall) — re-creating it here is what brings the waiter
 734        // back from "permanently invisible". Same document shape as UpsertSubscriberAsync.
 735        var writes = new List<WriteModel<MongoChannelSubscriberDocument>>(registrations.Count);
 736        foreach (var (correlationId, registrationId) in registrations)
 737        {
 738            writes.Add(new UpdateOneModel<MongoChannelSubscriberDocument>(
 739                Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, regis
 740                BuildSubscriberUpsertPipeline(correlationId, registrationId, instanceId, ttl))
 741            {
 742                IsUpsert = true
 743            });
 744        }
 745
 746        await _subscribers.BulkWriteAsync(
 747            writes,
 748            new BulkWriteOptions { IsOrdered = false },
 749            cancellationToken).ConfigureAwait(false);
 750    }
 751
 752    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 753    {
 754        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 755        await _subscribers.DeleteOneAsync(
 756            Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.Id, RegistrationKey(correlationId, registrat
 757            cancellationToken).ConfigureAwait(false);
 758    }
 759
 760    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 761    {
 762        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 763        var filter = Builders<MongoChannelSubscriberDocument>.Filter.Eq(item => item.CorrelationId, correlationId)
 764                     & NotExpiredOnServerClock<MongoChannelSubscriberDocument>();
 765        return await _subscribers.CountDocumentsAsync(filter, cancellationToken: cancellationToken).ConfigureAwait(false
 766    }
 767
 768    /// <summary>
 769    /// Server-clock expiry filter (<c>$expr: expires_at &gt; $$NOW</c>): message, liveness, and
 770    /// recovery expiry are all stamped with $$NOW, so comparing them against the app clock would
 771    /// reintroduce the clock-skew hole the server-side stamps exist to close.
 772    /// </summary>
 773    internal static FilterDefinition<TDocument> NotExpiredOnServerClock<TDocument>()
 774        => new BsonDocument("$expr", new BsonDocument("$gt", new BsonArray { "$expires_at", "$$NOW" }));
 775
 776    /// <summary>
 777    /// Watches the message collection with a change stream and invokes
 778    /// <paramref name="onNotification"/> with the correlation id of every inserted response. One
 779    /// stream serves every local waiter — the caller routes the id to the right subscription — so
 780    /// waiter count never multiplies server-side cursors. Runs until cancellation or a stream error.
 781    /// </summary>
 782    public async Task WatchMessagesAsync(Func<string?, Task> onNotification, CancellationToken cancellationToken)
 783    {
 784        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 785        using var cursor = await _messages.WatchAsync(
 786            BuildMessageWatchPipeline(),
 787            new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup },
 788            cancellationToken).ConfigureAwait(false);
 789        while (await cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false))
 790        {
 791            foreach (var change in cursor.Current)
 792                await onNotification(change.FullDocument?.CorrelationId).ConfigureAwait(false);
 793        }
 794    }
 795
 796    /// <summary>
 797    /// Change-stream pipeline for response wakes: a <c>$match</c> on insert events. The correlation
 798    /// id travels in the event's full document, letting the dispatcher scan only the signaled
 799    /// correlation id — the targeted-wake contract.
 800    /// </summary>
 801    internal static PipelineDefinition<ChangeStreamDocument<MongoChannelMessageDocument>, ChangeStreamDocument<MongoChan
 802        => new EmptyPipelineDefinition<ChangeStreamDocument<MongoChannelMessageDocument>>()
 803            .Match(change => change.OperationType == ChangeStreamOperationType.Insert);
 804
 805    /// <summary>Returns <c>true</c> when the server rejected the change stream itself (not a transient cursor error).</
 806    internal static bool IsChangeStreamUnsupported(Exception exception)
 807        => exception is MongoCommandException commandException
 808           && (commandException.Code == 40573
 809               || commandException.Message.Contains("only supported on replica sets", StringComparison.OrdinalIgnoreCase
 810
 811    internal static string RegistrationKey(string correlationId, Guid registrationId)
 812        => $"{correlationId}:{registrationId:N}";
 813
 814    /// <summary>
 815    /// Name of the derived ack-counter collection. Part of the effective collection-name plan:
 816    /// options validation must keep the configured collections distinct from this derived name,
 817    /// or counter documents land in (for example) the TTL-indexed recovery collection, where the
 818    /// reaper would silently delete the ack sequence and reset the same-tick tie-breaker.
 819    /// </summary>
 820    internal static string CountersCollectionName(string messageCollection) => $"{messageCollection}_counters";
 821
 822    /// <summary>Validates a MongoDB collection name coming from options.</summary>
 823    public static void ValidateCollectionName(string? value, string name)
 824    {
 825        if (string.IsNullOrWhiteSpace(value))
 826            throw new InvalidOperationException($"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} must be configured
 827        if (value.Contains('$') || value.Contains('\0')
 828            || value.StartsWith("system.", StringComparison.Ordinal) || value.Contains(".system.", StringComparison.Ordi
 829            throw new InvalidOperationException(
 830                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} '{value}' must be a valid MongoDB collection name 
 831        if (string.Equals(value, MongoOwnershipLedger.CollectionName, StringComparison.Ordinal))
 832            throw new InvalidOperationException(
 833                $"{nameof(MongoDbAsyncResponseChannelOptions)}.{name} '{value}' is reserved for the cross-component owne
 834    }
 835
 836    /// <summary>Disposes the Mongo client when the store created (and therefore owns) it.</summary>
 837    public void Dispose()
 838    {
 839        _ensureGate.Dispose();
 840        (_ownedClient as IDisposable)?.Dispose();
 841    }
 842}
 843
 844internal sealed class MongoRecoveryStateDocument
 845{
 846    [BsonId]
 847    [BsonElement("_id")]
 848    public string Id { get; set; } = "";
 849
 850    [BsonElement("correlation_id")]
 851    public string CorrelationId { get; set; } = "";
 852
 853    [BsonElement("registration_id")]
 854    [BsonGuidRepresentation(GuidRepresentation.Standard)]
 855    public Guid RegistrationId { get; set; }
 856
 857    [BsonElement("state_json")]
 858    public string StateJson { get; set; } = "";
 859
 860    [BsonElement("expires_at")]
 861    public DateTime ExpiresAtUtc { get; set; }
 862
 863    [BsonElement("registered_at")]
 864    public DateTime RegisteredAtUtc { get; set; }
 865}
 866
 867internal sealed class MongoChannelMessageDocument
 868{
 869    [BsonId]
 870    [BsonElement("_id")]
 871    [BsonGuidRepresentation(GuidRepresentation.Standard)]
 872    public Guid Id { get; set; }
 873
 874    [BsonElement("correlation_id")]
 875    public string CorrelationId { get; set; } = "";
 876
 877    /// <summary>Null only on a sweep projection of an acknowledged document (<see cref="MongoDbChannelStore.SweepProjec
 878    [BsonElement("envelope_json")]
 879    public string? EnvelopeJson { get; set; } = "";
 880
 881    [BsonElement("created_at")]
 882    public DateTime CreatedAtUtc { get; set; }
 883
 884    [BsonElement("expires_at")]
 885    public DateTime ExpiresAtUtc { get; set; }
 886
 887    [BsonElement("acked_at")]
 888    public DateTime? AckedAtUtc { get; set; }
 889
 890    [BsonElement("acked_seq")]
 891    public long? AckedSeq { get; set; }
 892
 893    [BsonElement("recovery_claimed")]
 894    public bool RecoveryClaimed { get; set; }
 895}
 896
 897internal sealed class MongoChannelSubscriberDocument
 898{
 899    [BsonId]
 900    [BsonElement("_id")]
 0901    public string Id { get; set; } = "";
 902
 903    [BsonElement("correlation_id")]
 0904    public string CorrelationId { get; set; } = "";
 905
 906    [BsonElement("registration_id")]
 907    [BsonGuidRepresentation(GuidRepresentation.Standard)]
 0908    public Guid RegistrationId { get; set; }
 909
 910    [BsonElement("instance_id")]
 0911    public string InstanceId { get; set; } = "";
 912
 913    [BsonElement("expires_at")]
 0914    public DateTime ExpiresAtUtc { get; set; }
 915}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

get_Id()
get_CorrelationId()
get_RegistrationId()
get_InstanceId()
get_ExpiresAtUtc()