CultLib/src/GameCult.Networking/CultNetDatabase.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using GameCult.Caching;
using R3;

namespace GameCult.Networking
{
    /// <summary>
    /// Describes the kind of domain change observed through a distributed CultCache surface.
    /// </summary>
    public enum CultNetDatabaseChangeKind
    {
        /// <summary>
        /// A document was added.
        /// </summary>
        Added,
        /// <summary>
        /// A document was updated.
        /// </summary>
        Updated,
        /// <summary>
        /// A document was removed.
        /// </summary>
        Removed,
        /// <summary>
        /// A document was accepted through compatible schema migration.
        /// </summary>
        SchemaMigrated,
        /// <summary>
        /// A document change was rejected.
        /// </summary>
        Rejected,
        /// <summary>
        /// A document was applied locally before authoritative commit.
        /// </summary>
        Predicted,
        /// <summary>
        /// A predicted document was replaced by the authoritative committed state.
        /// </summary>
        Reconciled
    }

    /// <summary>
    /// The one schema-alias matcher for every runtime-side schema match (docs/cultnet-selection-cut.md, D4).
    /// </summary>
    public static class CultNetSchemaAliasMatching
    {
        public static bool MatchesAny(IReadOnlyList<string> candidates, string schemaId)
        {
            return candidates.Count == 0 ||
                   candidates.Any(candidate => Matches(candidate, schemaId));
        }

        public static bool MatchesAny(IReadOnlyList<string> candidates, CultDocumentDescriptor descriptor)
        {
            return candidates.Count == 0 ||
                   candidates.Any(candidate => Matches(candidate, descriptor));
        }

        public static bool Matches(string candidate, string schemaId)
        {
            if (string.Equals(candidate, schemaId, StringComparison.Ordinal))
                return true;

            var candidateName = InferSchemaName(candidate) ?? candidate;
            var schemaName = InferSchemaName(schemaId) ?? schemaId;
            return string.Equals(candidateName, schemaName, StringComparison.Ordinal);
        }

        public static bool Matches(string candidate, CultDocumentDescriptor descriptor)
        {
            if (string.Equals(candidate, descriptor.SchemaId, StringComparison.Ordinal) ||
                string.Equals(candidate, descriptor.SchemaName, StringComparison.Ordinal) ||
                string.Equals(candidate, descriptor.SchemaVersion, StringComparison.Ordinal))
                return true;

            if (descriptor.ToCatalogEntry().CompatibleSchemaIds.Any(compatible =>
                    string.Equals(candidate, compatible, StringComparison.Ordinal)))
                return true;

            var candidateName = InferSchemaName(candidate) ?? candidate;
            return string.Equals(candidateName, descriptor.SchemaName, StringComparison.Ordinal);
        }

        private static string? InferSchemaName(string schemaId)
        {
            var marker = schemaId.LastIndexOf(".v", StringComparison.Ordinal);
            if (marker <= 0 || marker + 2 >= schemaId.Length)
                return null;

            var version = schemaId.Substring(marker + 2);
            return version.All(char.IsDigit)
                ? schemaId.Substring(0, marker)
                : null;
        }

        /// <summary>
        /// A selection whose schema filter also matches a document's wire binding id, when it carries
        /// one different from its descriptor's own. <see cref="CultDocumentDescriptor"/> knows nothing
        /// of <c>CultNetDocumentBinding</c>'s optional schema-id override, so a caller that has both in
        /// hand (the two subscription servers) reconciles them here rather than in the evaluator, which
        /// stays decoupled from CultNetDocumentRegistry by design.
        /// </summary>
        public static CultNetSelection WithBindingSchemaAlias(CultNetSelection selection, CultDocumentDescriptor descriptor, string? bindingSchemaId)
        {
            if (selection.Schemas is not { Length: > 0 } ||
                string.IsNullOrEmpty(bindingSchemaId) ||
                !selection.Schemas.Contains(bindingSchemaId, StringComparer.Ordinal) ||
                MatchesAny(selection.Schemas, descriptor))
            {
                return selection;
            }

            return new CultNetSelection
            {
                Schemas = null,
                Keys = selection.Keys,
                Fields = selection.Fields,
                Cites = selection.Cites,
                Cited = selection.Cited,
                Projection = selection.Projection,
                Descending = selection.Descending,
                Limit = selection.Limit,
                Cursor = selection.Cursor
            };
        }
    }

    /// <summary>
    /// Declares which documents this runtime may predict locally before authoritative commit.
    /// </summary>
    public sealed class CultNetClientAuthorityScope
    {
        /// <summary>
        /// Creates a client authority scope.
        /// </summary>
        public CultNetClientAuthorityScope(
            string ownerRuntimeId,
            IEnumerable<string>? schemaIds = null,
            string? keyPrefix = null)
        {
            OwnerRuntimeId = string.IsNullOrWhiteSpace(ownerRuntimeId)
                ? throw new ArgumentException("Value must be non-empty.", nameof(ownerRuntimeId))
                : ownerRuntimeId;
            SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray()
                ?? Array.Empty<string>();
            KeyPrefix = keyPrefix;
        }

        /// <summary>
        /// Gets the runtime that owns authoring for this scope.
        /// </summary>
        public string OwnerRuntimeId { get; }
        /// <summary>
        /// Gets the schema ids governed by this scope. Empty means all schemas.
        /// </summary>
        public IReadOnlyList<string> SchemaIds { get; }
        /// <summary>
        /// Gets the optional record-key prefix governed by this scope.
        /// </summary>
        public string? KeyPrefix { get; }

        internal bool Matches(string runtimeId, string schemaId, CultRecordKey key)
        {
            return string.Equals(OwnerRuntimeId, runtimeId, StringComparison.Ordinal) &&
                   CultNetSchemaAliasMatching.MatchesAny(SchemaIds, schemaId) &&
                   (string.IsNullOrEmpty(KeyPrefix) || key.Value.StartsWith(KeyPrefix!, StringComparison.Ordinal));
        }

        internal bool Matches(string runtimeId, CultDocumentDescriptor descriptor, CultRecordKey key)
        {
            return string.Equals(OwnerRuntimeId, runtimeId, StringComparison.Ordinal) &&
                   CultNetSchemaAliasMatching.MatchesAny(SchemaIds, descriptor) &&
                   (string.IsNullOrEmpty(KeyPrefix) || key.Value.StartsWith(KeyPrefix!, StringComparison.Ordinal));
        }
    }

    /// <summary>
    /// Describes one shard owned or observed by a CultNet database surface.
    /// </summary>
    public sealed class CultNetShardDescriptor
    {
        /// <summary>
        /// Creates a shard descriptor.
        /// </summary>
        public CultNetShardDescriptor(
            string shardId,
            string ownerRuntimeId,
            long epoch,
            bool isPrimary,
            IEnumerable<string>? schemaIds = null,
            string? keyPrefix = null,
            IEnumerable<string>? primaryEndpoints = null,
            IEnumerable<string>? replicaEndpoints = null,
            IEnumerable<string>? readReplicaEndpoints = null,
            string? region = null)
        {
            ShardId = RequireNonEmpty(shardId, nameof(shardId));
            OwnerRuntimeId = RequireNonEmpty(ownerRuntimeId, nameof(ownerRuntimeId));
            Epoch = epoch;
            IsPrimary = isPrimary;
            SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray()
                ?? Array.Empty<string>();
            KeyPrefix = keyPrefix;
            PrimaryEndpoints = CleanEndpoints(primaryEndpoints);
            ReplicaEndpoints = CleanEndpoints(replicaEndpoints);
            ReadReplicaEndpoints = CleanEndpoints(readReplicaEndpoints);
            Region = region;
        }

        /// <summary>
        /// Gets the stable shard identifier.
        /// </summary>
        public string ShardId { get; }
        /// <summary>
        /// Gets the runtime that currently owns writes for this shard.
        /// </summary>
        public string OwnerRuntimeId { get; }
        /// <summary>
        /// Gets the monotonic shard authority epoch.
        /// </summary>
        public long Epoch { get; }
        /// <summary>
        /// Gets whether the local database instance is authoritative for writes to this shard.
        /// </summary>
        public bool IsPrimary { get; }
        /// <summary>
        /// Gets the schema ids governed by this shard. Empty means all schemas.
        /// </summary>
        public IReadOnlyList<string> SchemaIds { get; }
        /// <summary>
        /// Gets the optional record-key prefix governed by this shard.
        /// </summary>
        public string? KeyPrefix { get; }
        /// <summary>
        /// Gets endpoints that can accept authoritative writes for this shard.
        /// </summary>
        public IReadOnlyList<string> PrimaryEndpoints { get; }
        /// <summary>
        /// Gets endpoints that replicate authoritative shard mutations.
        /// </summary>
        public IReadOnlyList<string> ReplicaEndpoints { get; }
        /// <summary>
        /// Gets endpoints intended for low-latency read and subscription traffic.
        /// </summary>
        public IReadOnlyList<string> ReadReplicaEndpoints { get; }
        /// <summary>
        /// Gets an optional locality label for this shard owner.
        /// </summary>
        public string? Region { get; }

        /// <summary>
        /// Creates a primary shard that accepts every schema and key.
        /// </summary>
        public static CultNetShardDescriptor PrimaryAll(string ownerRuntimeId = "local", string shardId = "primary")
        {
            return new CultNetShardDescriptor(shardId, ownerRuntimeId, epoch: 1, isPrimary: true);
        }

        /// <summary>
        /// Creates a read-only shard descriptor.
        /// </summary>
        public static CultNetShardDescriptor ReadOnly(
            string shardId,
            string ownerRuntimeId,
            long epoch = 1,
            IEnumerable<string>? schemaIds = null,
            string? keyPrefix = null)
        {
            return new CultNetShardDescriptor(shardId, ownerRuntimeId, epoch, isPrimary: false, schemaIds, keyPrefix);
        }

        internal bool Matches(string schemaId, CultRecordKey key)
        {
            var schemaMatches = CultNetSchemaAliasMatching.MatchesAny(SchemaIds, schemaId);
            var keyMatches = string.IsNullOrEmpty(KeyPrefix) ||
                             key.Value.StartsWith(KeyPrefix!, StringComparison.Ordinal);
            return schemaMatches && keyMatches;
        }

        internal bool Matches(CultDocumentDescriptor descriptor, CultRecordKey key)
        {
            var schemaMatches = CultNetSchemaAliasMatching.MatchesAny(SchemaIds, descriptor);
            var keyMatches = string.IsNullOrEmpty(KeyPrefix) ||
                             key.Value.StartsWith(KeyPrefix!, StringComparison.Ordinal);
            return schemaMatches && keyMatches;
        }

        private static string RequireNonEmpty(string value, string paramName)
        {
            if (string.IsNullOrWhiteSpace(value))
            {
                throw new ArgumentException("Value must be non-empty.", paramName);
            }

            return value;
        }

        private static string[] CleanEndpoints(IEnumerable<string>? endpoints)
        {
            return endpoints?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray()
                ?? Array.Empty<string>();
        }
    }

    /// <summary>
    /// Options for creating a distributed CultCache database surface.
    /// </summary>
    public sealed class CultNetDatabaseOptions
    {
        /// <summary>
        /// Gets or sets the local runtime id advertised as shard owner for default primary shards.
        /// </summary>
        public string RuntimeId { get; set; } = "local";
        /// <summary>
        /// Gets or sets explicit shard descriptors. When omitted, this instance owns one primary shard for all records.
        /// </summary>
        public IReadOnlyList<CultNetShardDescriptor>? Shards { get; set; }
        /// <summary>
        /// Gets or sets document scopes this runtime may predict before authoritative commit.
        /// </summary>
        public IReadOnlyList<CultNetClientAuthorityScope>? ClientAuthorityScopes { get; set; }
        /// <summary>
        /// Gets or sets the document registry used to create and apply raw CultNet document messages.
        /// </summary>
        public CultNetDocumentRegistry? DocumentRegistry { get; set; }
        /// <summary>
        /// Gets or sets the durable store for accepted shard mutation logs.
        /// </summary>
        public ICultNetShardMutationLogStore? MutationLogStore { get; set; }
    }

    /// <summary>
    /// A typed domain change emitted by a CultNet database surface.
    /// </summary>
    public sealed class CultNetDatabaseChange<T> where T : class
    {
        /// <summary>
        /// Creates a database change.
        /// </summary>
        public CultNetDatabaseChange(
            CultNetDatabaseChangeKind kind,
            CultRecordKey key,
            string schemaId,
            CultNetShardDescriptor shard,
            T? document,
            T? previousDocument,
            string? message = null)
        {
            Kind = kind;
            Key = key;
            SchemaId = schemaId;
            Shard = shard;
            Document = document;
            PreviousDocument = previousDocument;
            Message = message;
        }

        /// <summary>
        /// Gets the change kind.
        /// </summary>
        public CultNetDatabaseChangeKind Kind { get; }
        /// <summary>
        /// Gets the changed record key.
        /// </summary>
        public CultRecordKey Key { get; }
        /// <summary>
        /// Gets the schema id of the changed document.
        /// </summary>
        public string SchemaId { get; }
        /// <summary>
        /// Gets the shard that accepted or rejected the change.
        /// </summary>
        public CultNetShardDescriptor Shard { get; }
        /// <summary>
        /// Gets the current document for added and updated events.
        /// </summary>
        public T? Document { get; }
        /// <summary>
        /// Gets the previous document for updated and removed events.
        /// </summary>
        public T? PreviousDocument { get; }
        /// <summary>
        /// Gets an optional diagnostic message.
        /// </summary>
        public string? Message { get; }
    }

    /// <summary>
    /// One ordered mutation accepted for a shard.
    /// </summary>
    public sealed class CultNetShardMutationLogEntry
    {
        /// <summary>
        /// Creates a shard mutation log entry.
        /// </summary>
        public CultNetShardMutationLogEntry(
            string shardId,
            long shardEpoch,
            long sequence,
            string committedAt,
            CultNetDatabaseChangeKind kind,
            string schemaId,
            CultRecordKey key,
            object? document,
            object? previousDocument)
        {
            ShardId = shardId;
            ShardEpoch = shardEpoch;
            Sequence = sequence;
            CommittedAt = committedAt;
            Kind = kind;
            SchemaId = schemaId;
            Key = key;
            Document = document;
            PreviousDocument = previousDocument;
        }

        /// <summary>
        /// Gets the shard id.
        /// </summary>
        public string ShardId { get; }
        /// <summary>
        /// Gets the shard epoch.
        /// </summary>
        public long ShardEpoch { get; }
        /// <summary>
        /// Gets the per-shard mutation sequence.
        /// </summary>
        public long Sequence { get; }
        /// <summary>
        /// Gets the commit timestamp.
        /// </summary>
        public string CommittedAt { get; }
        /// <summary>
        /// Gets the mutation kind.
        /// </summary>
        public CultNetDatabaseChangeKind Kind { get; }
        /// <summary>
        /// Gets the schema id.
        /// </summary>
        public string SchemaId { get; }
        /// <summary>
        /// Gets the record key.
        /// </summary>
        public CultRecordKey Key { get; }
        /// <summary>
        /// Gets the current document, when present.
        /// </summary>
        public object? Document { get; }
        /// <summary>
        /// Gets the previous document, when present.
        /// </summary>
        public object? PreviousDocument { get; }
    }

    /// <summary>
    /// Raised when a database write targets a shard this node does not own.
    /// </summary>
    public sealed class CultNetShardAuthorityException : InvalidOperationException
    {
        /// <summary>
        /// Creates a shard authority exception.
        /// </summary>
        public CultNetShardAuthorityException(CultNetShardDescriptor shard, string message, string reason = "not_primary") : base(message)
        {
            Shard = shard;
            Reason = reason;
        }

        /// <summary>
        /// Gets the shard that rejected the write.
        /// </summary>
        public CultNetShardDescriptor Shard { get; }
        /// <summary>
        /// Gets the machine-readable authority rejection reason.
        /// </summary>
        public string Reason { get; }
    }

    /// <summary>
    /// Database-style CultNet facade over a CultCache shard set.
    /// </summary>
    public sealed class CultNetDatabase : IDisposable
    {
        private readonly CultCache _cache;
        private readonly CultNetDocumentRegistry _documents;
        private readonly string _runtimeId;
        private readonly List<CultNetShardDescriptor> _shards;
        private readonly List<CultNetClientAuthorityScope> _clientAuthorityScopes;
        private readonly ICultNetShardMutationLogStore? _mutationLogStore;
        private readonly Dictionary<string, object> _predictedDocuments = new(StringComparer.Ordinal);
        private readonly Dictionary<string, List<CultNetShardMutationLogEntry>> _mutationLogs =
            new(StringComparer.Ordinal);
        private readonly Dictionary<string, long> _nextLogSequences = new(StringComparer.Ordinal);
        private readonly Dictionary<string, long> _appliedShardSequences = new(StringComparer.Ordinal);
        // CultNet typed selection, section 2: "ordinal is the sequence of the commit that last wrote
        // the row". Kept on every append (AppendMutationLogEntry is the one site), keyed by (schemaId,
        // recordKey) so the evaluator's order and cursor never read a clock.
        private readonly Dictionary<(string SchemaId, string RecordKey), long> _lastWriteSequence =
            new();
        private readonly Subject<object> _changes = new();
        private bool _disposed;

        /// <summary>
        /// Creates a database surface over a CultCache instance.
        /// </summary>
        public CultNetDatabase(CultCache cache, CultNetDatabaseOptions? options = null)
        {
            _cache = cache ?? throw new ArgumentNullException(nameof(cache));
            options ??= new CultNetDatabaseOptions();
            _runtimeId = options.RuntimeId;
            _documents = options.DocumentRegistry ?? new CultNetDocumentRegistry(cache.Registry);
            _mutationLogStore = options.MutationLogStore;
            _shards = (options.Shards == null || options.Shards.Count == 0
                    ? new[] { CultNetShardDescriptor.PrimaryAll(options.RuntimeId) }
                    : options.Shards)
                .ToList();
            _clientAuthorityScopes = (options.ClientAuthorityScopes ?? Array.Empty<CultNetClientAuthorityScope>()).ToList();
            InitializeLogSequencesFromStore();
            _cache.OnUpdate += PublishCacheUpdate;
        }

        /// <summary>
        /// Gets the local cache backing this database surface.
        /// </summary>
        public CultCache Cache => _cache;
        /// <summary>
        /// Gets the document registry used by the raw CultNet lane.
        /// </summary>
        public CultNetDocumentRegistry Documents => _documents;
        /// <summary>
        /// Gets the shard map known to this database surface.
        /// </summary>
        public IReadOnlyList<CultNetShardDescriptor> Shards => _shards;

        /// <summary>
        /// Gets the document scopes this runtime may predict before authoritative commit.
        /// </summary>
        public IReadOnlyList<CultNetClientAuthorityScope> ClientAuthorityScopes => _clientAuthorityScopes;

        /// <summary>Gets whether this database instance can authoritatively replace the document at a key.</summary>
        public bool CanWriteAuthoritatively<T>(CultRecordKey key) where T : class
        {
            ThrowIfDisposed();
            var descriptor = _cache.Registry.GetRequired<T>();
            return ResolveShardInternal(descriptor, key).IsPrimary;
        }

        /// <summary>Gets whether this database instance can submit a prediction for the document at a key.</summary>
        public bool CanSubmitPrediction<T>(CultRecordKey key) where T : class
        {
            ThrowIfDisposed();
            var descriptor = _cache.Registry.GetRequired<T>();
            return _clientAuthorityScopes.Any(scope => scope.Matches(_runtimeId, descriptor, key));
        }

        /// <summary>
        /// Gets mutation log entries for a shard after the supplied sequence.
        /// </summary>
        public IReadOnlyList<CultNetShardMutationLogEntry> GetMutationLog(
            string shardId,
            long afterSequence = 0,
            int? limit = null)
        {
            ThrowIfDisposed();
            if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId));
            if (!_mutationLogs.TryGetValue(shardId, out var entries))
            {
                return Array.Empty<CultNetShardMutationLogEntry>();
            }

            var query = entries.Where(entry => entry.Sequence > afterSequence);
            if (limit.HasValue)
            {
                query = query.Take(limit.Value);
            }

            return query.ToArray();
        }

        /// <summary>
        /// Gets replica catch-up log entries for a shard after the supplied sequence.
        /// </summary>
        public IReadOnlyList<CultNetShardLogEntryMessage> GetMutationLogMessages(
            string shardId,
            long afterSequence = 0,
            int? limit = null)
        {
            ThrowIfDisposed();
            if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId));
            if (_mutationLogStore != null)
            {
                return _mutationLogStore.Read(shardId, afterSequence, limit);
            }

            return GetMutationLog(shardId, afterSequence, limit)
                .Select(ToLogEntryMessage)
                .ToArray();
        }

        /// <summary>
        /// Gets the highest retained-log sequence that has been compacted away for a shard.
        /// </summary>
        public long GetCompactedMutationLogSequence(string shardId)
        {
            ThrowIfDisposed();
            if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId));
            return _mutationLogStore?.GetCompactedThrough(shardId) ?? 0;
        }

        /// <summary>
        /// Gets the latest known shard-log sequence for a shard.
        /// </summary>
        public long GetLatestMutationLogSequence(string shardId)
        {
            ThrowIfDisposed();
            if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId));
            var compactedThrough = GetCompactedMutationLogSequence(shardId);
            var retained = _mutationLogStore != null
                ? _mutationLogStore.Read(shardId).Select(entry => entry.Sequence)
                : GetMutationLog(shardId).Select(entry => entry.Sequence);
            return retained.DefaultIfEmpty(compactedThrough).Max();
        }

        /// <summary>
        /// Creates a raw document snapshot bounded to one shard.
        /// </summary>
        public CultNetSnapshotResponseRawMessage CreateShardSnapshotResponse(
            CultNetShardDescriptor shard,
            string messageId,
            CultNetSnapshotRequestMessage? filter = null)
        {
            ThrowIfDisposed();
            if (shard == null) throw new ArgumentNullException(nameof(shard));

            var requestedSchemaIds = filter?.SchemaIds;
            var requestedRecordKeys = filter?.RecordKeys != null
                ? new HashSet<string>(filter.RecordKeys, StringComparer.Ordinal)
                : null;
            var documents = new List<CultNetRawDocumentRecord>();
            foreach (var document in _cache.AllEntries)
            {
                var documentType = document.GetType();
                var descriptor = _cache.Registry.GetRequired(documentType);
                var key = GetTrackedKey(document, documentType);
                if (string.IsNullOrWhiteSpace(key.Value) ||
                    !shard.Matches(descriptor.SchemaId, key) ||
                    (requestedSchemaIds != null && !CultNetSchemaAliasMatching.MatchesAny(requestedSchemaIds, descriptor)) ||
                    (requestedRecordKeys != null && !requestedRecordKeys.Contains(key.Value)))
                {
                    continue;
                }

                documents.Add(CreateRawDocumentRecord(key, document));
            }

            return new CultNetSnapshotResponseRawMessage
            {
                MessageId = string.IsNullOrWhiteSpace(messageId) ? Guid.NewGuid().ToString("N") : messageId,
                Documents = documents.ToArray(),
                ShardId = shard.ShardId,
                ShardEpoch = shard.Epoch,
                ShardLogSequence = GetLatestMutationLogSequence(shard.ShardId)
            };
        }

        /// <summary>
        /// Applies a shard-bounded snapshot and advances the replica cursor to its represented sequence.
        /// </summary>
        public async Task<long> ApplyShardSnapshotResponseAsync(
            CultNetShardDescriptor shard,
            CultNetSnapshotResponseRawMessage response)
        {
            ThrowIfDisposed();
            if (shard == null) throw new ArgumentNullException(nameof(shard));
            if (response == null) throw new ArgumentNullException(nameof(response));
            if (!string.IsNullOrWhiteSpace(response.ShardId) &&
                !string.Equals(response.ShardId, shard.ShardId, StringComparison.Ordinal))
            {
                throw new InvalidOperationException($"Snapshot for shard '{response.ShardId}' cannot be applied to shard '{shard.ShardId}'.");
            }

            if (response.ShardEpoch.HasValue && response.ShardEpoch.Value != shard.Epoch)
            {
                throw new CultNetShardAuthorityException(
                    shard,
                    $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not snapshot epoch {response.ShardEpoch.Value}.",
                    "stale_epoch");
            }

            var incomingKeys = response.Documents
                .Where(document => ShardMatchesRawDocument(shard, document))
                .Select(document => document.RecordKey)
                .ToHashSet(StringComparer.Ordinal);
            var local = GetLocalShardDocuments(shard).ToArray();
            foreach (var existing in local.Where(item => !incomingKeys.Contains(item.Key.Value)))
            {
                RemoveTrackedDocument(existing.DocumentType, existing.Key);
                PublishUntyped(
                    existing.DocumentType,
                    CultNetDatabaseChangeKind.Removed,
                    existing.Key,
                    existing.SchemaId,
                    shard,
                    document: null,
                    previousDocument: existing.Document);
            }

            foreach (var document in response.Documents.Where(document => ShardMatchesRawDocument(shard, document)))
            {
                var key = new CultRecordKey(document.RecordKey);
                var descriptor = _documents.ResolveDescriptorForRawDocument(document);
                var previous = _cache.Get(key);
                var applied = await _documents.ApplyRawDocumentPutMessageAsync(
                    _cache,
                    new CultNetDocumentPutRawMessage
                    {
                        MessageId = response.MessageId,
                        Document = document,
                        ShardId = shard.ShardId,
                        ShardEpoch = shard.Epoch
                    }).ConfigureAwait(false);
                PublishUntyped(
                    descriptor.DocumentType,
                    previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated,
                    key,
                    descriptor.SchemaId,
                    shard,
                    applied,
                    previous);
            }

            var sequence = response.ShardLogSequence ?? 0;
            SetAppliedShardSequence(shard.ShardId, sequence);
            return sequence;
        }

        /// <summary>
        /// Gets the last replicated sequence applied for a shard.
        /// </summary>
        public long GetAppliedShardSequence(string shardId)
        {
            ThrowIfDisposed();
            if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId));
            return _appliedShardSequences.TryGetValue(shardId, out var sequence)
                ? sequence
                : 0;
        }

        internal void SetAppliedShardSequence(string shardId, long sequence)
        {
            ThrowIfDisposed();
            if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId));
            if (sequence < 0) throw new ArgumentOutOfRangeException(nameof(sequence), "Sequence must be non-negative.");
            _appliedShardSequences[shardId] = sequence;
        }

        /// <summary>
        /// Creates a shard catalog response for optional schema/key filters.
        /// </summary>
        public CultNetShardCatalogResponseMessage CreateShardCatalogResponse(CultNetShardCatalogRequestMessage request)
        {
            ThrowIfDisposed();
            if (request == null) throw new ArgumentNullException(nameof(request));

            var schemaIds = request.SchemaIds == null || request.SchemaIds.Length == 0
                ? null
                : request.SchemaIds;
            var recordKeys = request.RecordKeys == null || request.RecordKeys.Length == 0
                ? null
                : request.RecordKeys.Select(key => new CultRecordKey(key)).ToArray();

            var descriptors = _shards
                .Where(shard => MatchesCatalogFilter(shard, schemaIds, recordKeys))
                .Select(ToMessage)
                .ToArray();

            return new CultNetShardCatalogResponseMessage
            {
                MessageId = string.IsNullOrWhiteSpace(request.MessageId)
                    ? Guid.NewGuid().ToString("N")
                    : request.MessageId,
                Shards = descriptors
            };
        }

        /// <summary>
        /// Resolves the shard that governs the supplied schema and key.
        /// </summary>
        public CultNetShardDescriptor ResolveShard(string schemaId, CultRecordKey key)
        {
            ThrowIfDisposed();
            return ResolveShardInternal(schemaId, key);
        }

        /// <summary>
        /// Gets a document by key.
        /// </summary>
        public Task<T?> GetAsync<T>(CultRecordKey key) where T : class
        {
            ThrowIfDisposed();
            return Task.FromResult(_cache.Get<T>(key));
        }

        /// <summary>
        /// Gets all documents assignable to the requested type.
        /// </summary>
        public IEnumerable<T> GetAll<T>() where T : class
        {
            ThrowIfDisposed();
            return _cache.GetAll<T>();
        }

        /// <summary>
        /// Gets a typed document by its CultName value.
        /// </summary>
        public T? GetByName<T>(string name) where T : class
        {
            ThrowIfDisposed();
            return _cache.GetByName<T>(name);
        }

        /// <summary>
        /// Gets a typed document by an indexed value.
        /// </summary>
        public T? GetByIndex<T>(string alias, string value) where T : class
        {
            ThrowIfDisposed();
            return _cache.GetByIndex<T>(alias, value);
        }

        /// <summary>
        /// Adds or replaces a document at a specific key.
        /// </summary>
        public async Task<CultRecordHandle<T>> PutAsync<T>(CultRecordKey key, T document) where T : class
        {
            ThrowIfDisposed();
            if (document == null) throw new ArgumentNullException(nameof(document));

            var descriptor = _cache.Registry.GetRequired(document.GetType());
            var shard = ResolveShardInternal(descriptor, key);
            EnsurePrimary(shard, descriptor.SchemaId, key);

            var previous = _cache.Get<T>(key);
            var handle = await _cache.UpsertAsync(document, new CultRecordHandle<T>(key)).ConfigureAwait(false);
            void PublishCommittedPut()
            {
                AppendMutationLogEntry(
                    shard,
                    previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated,
                    key,
                    descriptor.SchemaId,
                    document,
                    previous);
                Publish(new CultNetDatabaseChange<T>(
                    previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated,
                    key,
                    descriptor.SchemaId,
                    shard,
                    document,
                    previous));
            }
            PublishCommittedPut();
            return handle;
        }

        /// <summary>
        /// Applies a locally predicted document for a client-owned input scope.
        /// </summary>
        public async Task<CultRecordHandle<T>> PutPredictedAsync<T>(CultRecordKey key, T document) where T : class
        {
            ThrowIfDisposed();
            if (document == null) throw new ArgumentNullException(nameof(document));

            var descriptor = _cache.Registry.GetRequired(document.GetType());
            var shard = ResolveShardInternal(descriptor, key);
            EnsureClientAuthority(descriptor, key);

            var previous = _cache.Get<T>(key);
            var handle = await _cache.UpsertAsync(document, new CultRecordHandle<T>(key)).ConfigureAwait(false);
            _predictedDocuments[PredictionKey(descriptor.SchemaId, key)] = document;
            Publish(new CultNetDatabaseChange<T>(
                CultNetDatabaseChangeKind.Predicted,
                key,
                descriptor.SchemaId,
                shard,
                document,
                previous));
            return handle;
        }

        /// <summary>
        /// Deletes a document by key.
        /// </summary>
        public Task DeleteAsync<T>(CultRecordKey key) where T : class
        {
            ThrowIfDisposed();
            var descriptor = _cache.Registry.GetRequired<T>();
            var shard = ResolveShardInternal(descriptor, key);
            EnsurePrimary(shard, descriptor.SchemaId, key);

            var previous = _cache.Get<T>(key);
            if (previous != null)
            {
                _cache.Remove(new CultRecordHandle<T>(key));
                void PublishCommittedDelete()
                {
                    AppendMutationLogEntry(
                        shard,
                        CultNetDatabaseChangeKind.Removed,
                        key,
                        descriptor.SchemaId,
                        document: null,
                        previousDocument: previous);
                    Publish(new CultNetDatabaseChange<T>(
                        CultNetDatabaseChangeKind.Removed,
                        key,
                        descriptor.SchemaId,
                        shard,
                        document: null,
                        previousDocument: previous));
                }
                PublishCommittedDelete();
            }

            return Task.CompletedTask;
        }

        /// <summary>
        /// Applies a raw document mutation after checking shard authority.
        /// </summary>
        public async Task<object> ApplyPutAsync(CultNetDocumentPutRawMessage message)
        {
            ThrowIfDisposed();
            if (message == null) throw new ArgumentNullException(nameof(message));
            if (message.Document == null)
            {
                throw new ArgumentException("CultNet raw document message is missing its document payload.", nameof(message));
            }

            var key = new CultRecordKey(message.Document.RecordKey);
            var descriptor = _documents.ResolveDescriptorForRawDocument(message.Document);
            var shard = ResolveShardInternal(descriptor, key);
            EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch);
            var previous = _cache.Get(key);
            var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false);
            var kind = previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated;
            var predictionKey = PredictionKey(descriptor.SchemaId, key);
            var reconcilesPrediction = _predictedDocuments.ContainsKey(predictionKey);
            if (reconcilesPrediction)
            {
                kind = CultNetDatabaseChangeKind.Reconciled;
            }

            void PublishCommittedPut()
            {
                if (reconcilesPrediction)
                    _predictedDocuments.Remove(predictionKey);
                AppendMutationLogEntry(
                    shard,
                    kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind,
                    key,
                    descriptor.SchemaId,
                    document,
                    previous,
                    new CultNetShardLogEntryMessage
                    {
                        ChangeKind = previous == null ? "added" : "updated",
                        Put = message
                    });
                PublishUntyped(
                    descriptor.DocumentType,
                    kind,
                    key,
                    descriptor.SchemaId,
                    shard,
                    document,
                    previous);
            }
            PublishCommittedPut();
            return document;
        }

        /// <summary>
        /// Applies a raw document delete after checking shard authority.
        /// </summary>
        public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message)
        {
            ThrowIfDisposed();
            if (message == null) throw new ArgumentNullException(nameof(message));

            var key = new CultRecordKey(message.RecordKey);
            var descriptor = _documents.ResolveDescriptorForSchemaId(message.SchemaId);
            var shard = ResolveShardInternal(descriptor, key);
            EnsurePrimary(shard, descriptor.SchemaId, key, message.ShardEpoch);
            var previous = _cache.Get(key);
            if (previous == null)
            {
                return Task.CompletedTask;
            }

            _cache.Remove(key);
            void PublishCommittedDelete()
            {
                AppendMutationLogEntry(
                    shard,
                    CultNetDatabaseChangeKind.Removed,
                    key,
                    descriptor.SchemaId,
                    document: null,
                    previousDocument: previous,
                    wireEntry: new CultNetShardLogEntryMessage
                    {
                        ChangeKind = "removed",
                        Delete = message
                    });
                PublishUntyped(
                    descriptor.DocumentType,
                    CultNetDatabaseChangeKind.Removed,
                    key,
                    descriptor.SchemaId,
                    shard,
                    document: null,
                    previousDocument: previous);
            }
            PublishCommittedDelete();
            return Task.CompletedTask;
        }

        /// <summary>
        /// Applies committed shard-log entries from an authoritative peer.
        /// </summary>
        public async Task<long> ApplyShardLogResponseAsync(CultNetShardLogResponseMessage response)
        {
            ThrowIfDisposed();
            if (response == null) throw new ArgumentNullException(nameof(response));
            if (response.ResyncRequired)
            {
                throw new InvalidOperationException(
                    $"Shard '{response.ShardId}' requires snapshot resync before log application: {response.Reason ?? "unspecified"}.");
            }

            var shard = _shards.FirstOrDefault(candidate =>
                string.Equals(candidate.ShardId, response.ShardId, StringComparison.Ordinal));
            if (shard == null)
            {
                throw new InvalidOperationException($"Shard '{response.ShardId}' is not known by this database.");
            }

            if (response.ShardEpoch != shard.Epoch)
            {
                throw new CultNetShardAuthorityException(
                    shard,
                    $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not response epoch {response.ShardEpoch}.",
                    "stale_epoch");
            }

            var lastApplied = GetAppliedShardSequence(response.ShardId);
            var orderedEntries = response.Entries.OrderBy(entry => entry.Sequence).ToArray();
            foreach (var entry in orderedEntries)
            {
                if (entry.Sequence <= lastApplied)
                {
                    continue;
                }

                if (entry.Sequence != lastApplied + 1)
                {
                    throw new InvalidOperationException(
                        $"Shard '{response.ShardId}' log has a gap. Expected sequence {lastApplied + 1}, received {entry.Sequence}.");
                }

                await ApplyCommittedShardLogEntryAsync(shard, entry).ConfigureAwait(false);
                lastApplied = entry.Sequence;
                _appliedShardSequences[response.ShardId] = lastApplied;
            }

            return lastApplied;
        }

        /// <summary>
        /// Watches all changes assignable to the requested document type.
        /// </summary>
        public Observable<CultNetDatabaseChange<T>> Watch<T>() where T : class
        {
            ThrowIfDisposed();
            return _changes
                .Where(change => change is CultNetDatabaseChange<T>)
                .Select(change => (CultNetDatabaseChange<T>)change);
        }

        /// <summary>
        /// Watches all typed domain changes as boxed change objects.
        /// </summary>
        public Observable<object> WatchAllChanges()
        {
            ThrowIfDisposed();
            return _changes;
        }

        /// <summary>
        /// Watches one record.
        /// </summary>
        public Observable<CultNetDatabaseChange<T>> WatchRecord<T>(CultRecordKey key) where T : class
        {
            return Watch<T>().Where(change => change.Key.Equals(key));
        }

        /// <summary>
        /// Watches the global record for a document type.
        /// </summary>
        public Observable<CultNetDatabaseChange<T>> WatchGlobal<T>() where T : class
        {
            var descriptor = _cache.Registry.GetRequired<T>();
            return WatchRecord<T>(new CultRecordKey($"global:{descriptor.SchemaId}"));
        }

        /// <summary>
        /// Watches changes for a named document.
        /// </summary>
        public Observable<CultNetDatabaseChange<T>> WatchByName<T>(string name) where T : class
        {
            if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("Value must be non-empty.", nameof(name));
            return Watch<T>().Where(change =>
                (change.Document != null && ReferenceEquals(_cache.GetByName<T>(name), change.Document)) ||
                (change.PreviousDocument != null && ReferenceEquals(_cache.GetByName<T>(name), change.PreviousDocument)));
        }

        /// <summary>
        /// Watches changes for the current document mapped by an indexed value.
        /// </summary>
        public Observable<CultNetDatabaseChange<T>> WatchByIndex<T>(string alias, string value) where T : class
        {
            if (string.IsNullOrWhiteSpace(alias)) throw new ArgumentException("Value must be non-empty.", nameof(alias));
            if (string.IsNullOrWhiteSpace(value)) throw new ArgumentException("Value must be non-empty.", nameof(value));
            return Watch<T>().Where(change =>
                (change.Document != null && ReferenceEquals(_cache.GetByIndex<T>(alias, value), change.Document)) ||
                (change.PreviousDocument != null && ReferenceEquals(_cache.GetByIndex<T>(alias, value), change.PreviousDocument)));
        }

        /// <summary>
        /// Releases database subscriptions.
        /// </summary>
        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;
            _cache.OnUpdate -= PublishCacheUpdate;
            _changes.Dispose();
        }

        private void PublishCacheUpdate(object? previous, object? current)
        {
            var document = current ?? previous;
            if (document == null)
            {
                return;
            }

            var documentType = document.GetType();
            var descriptor = _cache.Registry.GetRequired(documentType);
            var key = current != null
                ? GetTrackedKey(current, documentType)
                : new CultRecordKey(string.Empty);
            var shard = ResolveShardInternal(descriptor, key);
            var changeType = typeof(CultNetDatabaseChange<>).MakeGenericType(documentType);
            var change = Activator.CreateInstance(
                changeType,
                current == null ? CultNetDatabaseChangeKind.Removed :
                previous == null ? CultNetDatabaseChangeKind.Added :
                CultNetDatabaseChangeKind.Updated,
                key,
                descriptor.SchemaId,
                shard,
                current,
                previous,
                null);
            if (change != null)
            {
                _changes.OnNext(change);
            }
        }

        private CultRecordKey GetTrackedKey(object document, Type documentType)
        {
            var method = typeof(CultCache).GetMethod(nameof(CultCache.TryGetHandle))!
                .MakeGenericMethod(documentType);
            var handleObject = method.Invoke(_cache, new[] { document });
            if (handleObject == null)
            {
                return new CultRecordKey(string.Empty);
            }

            var keyProperty = handleObject.GetType().GetProperty("Key");
            return keyProperty?.GetValue(handleObject) is CultRecordKey key
                ? key
                : new CultRecordKey(string.Empty);
        }

        private IEnumerable<TrackedShardDocument> GetLocalShardDocuments(CultNetShardDescriptor shard)
        {
            foreach (var document in _cache.AllEntries)
            {
                var documentType = document.GetType();
                var descriptor = _cache.Registry.GetRequired(documentType);
                var key = GetTrackedKey(document, documentType);
                if (!string.IsNullOrWhiteSpace(key.Value) && shard.Matches(descriptor, key))
                {
                    yield return new TrackedShardDocument(documentType, descriptor.SchemaId, key, document);
                }
            }
        }

        private bool ShardMatchesRawDocument(CultNetShardDescriptor shard, CultNetRawDocumentRecord document)
        {
            if (document == null ||
                string.IsNullOrWhiteSpace(document.SchemaId) ||
                string.IsNullOrWhiteSpace(document.RecordKey))
            {
                return false;
            }

            try
            {
                var descriptor = _documents.ResolveDescriptorForRawDocument(document);
                return shard.Matches(descriptor, new CultRecordKey(document.RecordKey));
            }
            catch (InvalidOperationException)
            {
                return false;
            }
        }

        private void RemoveTrackedDocument(Type documentType, CultRecordKey key)
        {
            _cache.Remove(key);
        }

        private void Publish<T>(CultNetDatabaseChange<T> change) where T : class
        {
            _changes.OnNext(change);
        }

        private sealed class TrackedShardDocument
        {
            public TrackedShardDocument(Type documentType, string schemaId, CultRecordKey key, object document)
            {
                DocumentType = documentType;
                SchemaId = schemaId;
                Key = key;
                Document = document;
            }

            public Type DocumentType { get; }
            public string SchemaId { get; }
            public CultRecordKey Key { get; }
            public object Document { get; }
        }

        private void AppendMutationLogEntry(
            CultNetShardDescriptor shard,
            CultNetDatabaseChangeKind kind,
            CultRecordKey key,
            string schemaId,
            object? document,
            object? previousDocument,
            CultNetShardLogEntryMessage? wireEntry = null)
        {
            var sequence = NextMutationLogSequence(shard.ShardId);
            var committedAt = DateTimeOffset.UtcNow.ToString("O");
            var entry = new CultNetShardMutationLogEntry(
                shard.ShardId,
                shard.Epoch,
                sequence,
                committedAt,
                kind,
                schemaId,
                key,
                document,
                previousDocument);
            var storedWireEntry = _mutationLogStore == null
                ? null
                : wireEntry ?? ToLogEntryMessage(entry);
            RecordMutationLogEntry(entry, storedWireEntry);
            _lastWriteSequence[(schemaId, key.Value)] = sequence;
        }

        /// <summary>
        /// The shard-log sequence of the commit that last wrote this row, or null when the row has
        /// never been committed through this database. The order the CultNet selection evaluator
        /// pages by (docs/cultnet-selection-cut.md, section 2).
        /// </summary>
        public long? LastWriteSequence(string schemaId, CultRecordKey key) =>
            _lastWriteSequence.TryGetValue((schemaId, key.Value), out var sequence) ? sequence : null;

        private void RecordMutationLogEntry(
            CultNetShardMutationLogEntry entry,
            CultNetShardLogEntryMessage? wireEntry = null)
        {
            if (!_mutationLogs.TryGetValue(entry.ShardId, out var entries))
            {
                entries = new List<CultNetShardMutationLogEntry>();
                _mutationLogs[entry.ShardId] = entries;
            }

            entries.Add(entry);
            if (!_nextLogSequences.TryGetValue(entry.ShardId, out var next) ||
                next <= entry.Sequence)
            {
                _nextLogSequences[entry.ShardId] = entry.Sequence + 1;
            }

            if (_mutationLogStore != null && wireEntry != null)
            {
                _mutationLogStore.Append(entry.ShardId, NormalizeWireLogEntry(entry, wireEntry));
            }
        }

        private long NextMutationLogSequence(string shardId)
        {
            var sequence = _nextLogSequences.TryGetValue(shardId, out var next)
                ? next
                : 1;
            _nextLogSequences[shardId] = sequence + 1;
            return sequence;
        }

        private void InitializeLogSequencesFromStore()
        {
            if (_mutationLogStore == null)
            {
                return;
            }

            foreach (var shard in _shards)
            {
                var highest = _mutationLogStore.Read(shard.ShardId)
                    .Select(entry => entry.Sequence)
                    .DefaultIfEmpty(0)
                    .Max();
                if (highest > 0)
                {
                    _nextLogSequences[shard.ShardId] = highest + 1;
                }
            }
        }

        private async Task ApplyCommittedShardLogEntryAsync(
            CultNetShardDescriptor shard,
            CultNetShardLogEntryMessage entry)
        {
            if (entry.Put != null)
            {
                await ApplyCommittedPutAsync(shard, entry).ConfigureAwait(false);
                return;
            }

            if (entry.Delete != null)
            {
                ApplyCommittedDelete(shard, entry);
                return;
            }

            throw new InvalidOperationException(
                $"Shard '{shard.ShardId}' log entry {entry.Sequence} has no put or delete payload.");
        }

        private async Task ApplyCommittedPutAsync(
            CultNetShardDescriptor shard,
            CultNetShardLogEntryMessage entry)
        {
            var message = entry.Put!;
            if (message.Document == null)
            {
                throw new InvalidOperationException(
                    $"Shard '{shard.ShardId}' log entry {entry.Sequence} put is missing its document payload.");
            }

            if (!string.Equals(message.ShardId, shard.ShardId, StringComparison.Ordinal) ||
                message.ShardEpoch != shard.Epoch)
            {
                throw new CultNetShardAuthorityException(
                    shard,
                    $"Shard '{shard.ShardId}' log entry {entry.Sequence} targets a different shard or epoch.",
                    "stale_epoch");
            }

            var key = new CultRecordKey(message.Document.RecordKey);
            var descriptor = _documents.ResolveDescriptorForRawDocument(message.Document);
            var previous = _cache.Get(key);
            var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false);
            var kind = ChangeKindFromWire(entry.ChangeKind);
            if (kind == CultNetDatabaseChangeKind.Removed)
            {
                throw new InvalidOperationException(
                    $"Shard '{shard.ShardId}' log entry {entry.Sequence} has a removed kind with a put payload.");
            }

            var predictionKey = PredictionKey(descriptor.SchemaId, key);
            if (_predictedDocuments.Remove(predictionKey))
            {
                kind = CultNetDatabaseChangeKind.Reconciled;
            }

            RecordMutationLogEntry(new CultNetShardMutationLogEntry(
                shard.ShardId,
                shard.Epoch,
                entry.Sequence,
                entry.CommittedAt,
                kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind,
                descriptor.SchemaId,
                key,
                document,
                previous),
                entry);
            PublishUntyped(
                descriptor.DocumentType,
                kind,
                key,
                descriptor.SchemaId,
                shard,
                document,
                previous);
        }

        private void ApplyCommittedDelete(
            CultNetShardDescriptor shard,
            CultNetShardLogEntryMessage entry)
        {
            var message = entry.Delete!;
            if (!string.Equals(message.ShardId, shard.ShardId, StringComparison.Ordinal) ||
                message.ShardEpoch != shard.Epoch)
            {
                throw new CultNetShardAuthorityException(
                    shard,
                    $"Shard '{shard.ShardId}' log entry {entry.Sequence} targets a different shard or epoch.",
                    "stale_epoch");
            }

            var key = new CultRecordKey(message.RecordKey);
            var descriptor = _cache.Registry.GetRequiredBySchemaId(message.SchemaId);
            var previous = _cache.Get(key);
            if (previous != null)
            {
                _cache.Remove(key);
            }

            RecordMutationLogEntry(new CultNetShardMutationLogEntry(
                shard.ShardId,
                shard.Epoch,
                entry.Sequence,
                entry.CommittedAt,
                CultNetDatabaseChangeKind.Removed,
                descriptor.SchemaId,
                key,
                document: null,
                previousDocument: previous),
                entry);
            PublishUntyped(
                descriptor.DocumentType,
                CultNetDatabaseChangeKind.Removed,
                key,
                descriptor.SchemaId,
                shard,
                document: null,
                previousDocument: previous);
        }

        private CultNetShardLogEntryMessage ToLogEntryMessage(CultNetShardMutationLogEntry entry)
        {
            if (entry.Kind == CultNetDatabaseChangeKind.Removed || entry.Document == null)
            {
                return new CultNetShardLogEntryMessage
                {
                    Sequence = entry.Sequence,
                    CommittedAt = entry.CommittedAt,
                    ChangeKind = "removed",
                    Delete = new CultNetDocumentDeleteMessage
                    {
                        MessageId = Guid.NewGuid().ToString("N"),
                        SchemaId = entry.SchemaId,
                        RecordKey = entry.Key.Value,
                        ShardId = entry.ShardId,
                        ShardEpoch = entry.ShardEpoch
                    }
                };
            }

            return new CultNetShardLogEntryMessage
            {
                Sequence = entry.Sequence,
                CommittedAt = entry.CommittedAt,
                ChangeKind = entry.Kind == CultNetDatabaseChangeKind.Added ? "added" : "updated",
                Put = new CultNetDocumentPutRawMessage
                {
                    MessageId = Guid.NewGuid().ToString("N"),
                    Document = CreateRawDocumentRecord(entry.Key, entry.Document),
                    ShardId = entry.ShardId,
                    ShardEpoch = entry.ShardEpoch
                }
            };
        }

        private CultNetShardLogEntryMessage NormalizeWireLogEntry(
            CultNetShardMutationLogEntry entry,
            CultNetShardLogEntryMessage wireEntry)
        {
            return new CultNetShardLogEntryMessage
            {
                Sequence = entry.Sequence,
                CommittedAt = string.IsNullOrWhiteSpace(entry.CommittedAt) ? wireEntry.CommittedAt : entry.CommittedAt,
                ChangeKind = entry.Kind == CultNetDatabaseChangeKind.Removed
                    ? "removed"
                    : entry.Kind == CultNetDatabaseChangeKind.Added ? "added" : "updated",
                Put = wireEntry.Put == null
                    ? null
                    : new CultNetDocumentPutRawMessage
                    {
                        MessageId = string.IsNullOrWhiteSpace(wireEntry.Put.MessageId)
                            ? Guid.NewGuid().ToString("N")
                            : wireEntry.Put.MessageId,
                        Document = wireEntry.Put.Document,
                        ShardId = entry.ShardId,
                        ShardEpoch = entry.ShardEpoch
                    },
                Delete = wireEntry.Delete == null
                    ? null
                    : new CultNetDocumentDeleteMessage
                    {
                        MessageId = string.IsNullOrWhiteSpace(wireEntry.Delete.MessageId)
                            ? Guid.NewGuid().ToString("N")
                            : wireEntry.Delete.MessageId,
                        SchemaId = wireEntry.Delete.SchemaId,
                        RecordKey = wireEntry.Delete.RecordKey,
                        ShardId = entry.ShardId,
                        ShardEpoch = entry.ShardEpoch
                    }
            };
        }

        private CultNetRawDocumentRecord CreateRawDocumentRecord(CultRecordKey key, object document)
        {
            var method = typeof(CultNetDocumentRegistry)
                .GetMethods(BindingFlags.Public | BindingFlags.Instance)
                .Single(candidate => candidate.Name == nameof(CultNetDocumentRegistry.CreateRawDocumentPutMessage) &&
                                     candidate.IsGenericMethodDefinition);
            var documentType = document.GetType();
            var handleType = typeof(CultRecordHandle<>).MakeGenericType(documentType);
            var handle = Activator.CreateInstance(handleType, new object[] { key });
            var message = method
                .MakeGenericMethod(documentType)
                .Invoke(_documents, new[] { Guid.NewGuid().ToString("N"), handle, document, null });
            return ((CultNetDocumentPutRawMessage)message!).Document;
        }

        private static CultNetDatabaseChangeKind ChangeKindFromWire(string changeKind)
        {
            return changeKind switch
            {
                "added" => CultNetDatabaseChangeKind.Added,
                "updated" => CultNetDatabaseChangeKind.Updated,
                "removed" => CultNetDatabaseChangeKind.Removed,
                _ => throw new InvalidOperationException($"Unsupported shard log change kind '{changeKind}'.")
            };
        }

        private void EnsureClientAuthority(string schemaId, CultRecordKey key)
        {
            if (_clientAuthorityScopes.Any(scope => scope.Matches(_runtimeId, schemaId, key)))
            {
                return;
            }

            var shard = ResolveShardInternal(schemaId, key);
            throw new CultNetShardAuthorityException(
                shard,
                $"Runtime '{_runtimeId}' does not have client prediction authority for schema '{schemaId}' key '{key.Value}'.",
                "not_client_authority");
        }

        private void EnsureClientAuthority(CultDocumentDescriptor descriptor, CultRecordKey key)
        {
            if (_clientAuthorityScopes.Any(scope => scope.Matches(_runtimeId, descriptor, key)))
            {
                return;
            }

            var shard = ResolveShardInternal(descriptor, key);
            throw new CultNetShardAuthorityException(
                shard,
                $"Runtime '{_runtimeId}' does not have client prediction authority for schema '{descriptor.SchemaId}' key '{key.Value}'.",
                "not_client_authority");
        }

        private static string PredictionKey(string schemaId, CultRecordKey key)
        {
            return $"{schemaId}:{key.Value}";
        }

        private void PublishUntyped(
            Type documentType,
            CultNetDatabaseChangeKind kind,
            CultRecordKey key,
            string schemaId,
            CultNetShardDescriptor shard,
            object? document,
            object? previousDocument)
        {
            var changeType = typeof(CultNetDatabaseChange<>).MakeGenericType(documentType);
            var change = Activator.CreateInstance(
                changeType,
                kind,
                key,
                schemaId,
                shard,
                document,
                previousDocument,
                null);
            if (change != null)
            {
                _changes.OnNext(change);
            }
        }

        private CultNetShardDescriptor ResolveShardInternal(string schemaId, CultRecordKey key)
        {
            return _shards.FirstOrDefault(shard => shard.Matches(schemaId, key)) ?? _shards[0];
        }

        private CultNetShardDescriptor ResolveShardInternal(CultDocumentDescriptor descriptor, CultRecordKey key)
        {
            return _shards.FirstOrDefault(shard => shard.Matches(descriptor, key)) ?? _shards[0];
        }

        private static void EnsurePrimary(CultNetShardDescriptor shard, string schemaId, CultRecordKey key, long? expectedEpoch = null)
        {
            if (expectedEpoch.HasValue && expectedEpoch.Value != shard.Epoch)
            {
                throw new CultNetShardAuthorityException(
                    shard,
                    $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not requested epoch {expectedEpoch.Value}.",
                    "stale_epoch");
            }

            if (shard.IsPrimary)
            {
                return;
            }

            throw new CultNetShardAuthorityException(
                shard,
                $"Shard '{shard.ShardId}' owned by '{shard.OwnerRuntimeId}' does not accept local writes for schema '{schemaId}' key '{key.Value}'.",
                "not_primary");
        }

        private bool MatchesCatalogFilter(
            CultNetShardDescriptor shard,
            IReadOnlyList<string>? schemaIds,
            IReadOnlyList<CultRecordKey>? recordKeys)
        {
            var schemaMatches = SchemaMatchesCatalogFilter(shard, schemaIds);
            var keyMatches = recordKeys == null ||
                             recordKeys.Count == 0 ||
                             recordKeys.Any(key => string.IsNullOrEmpty(shard.KeyPrefix) ||
                                                   key.Value.StartsWith(shard.KeyPrefix!, StringComparison.Ordinal));
            return schemaMatches && keyMatches;
        }

        private bool SchemaMatchesCatalogFilter(
            CultNetShardDescriptor shard,
            IReadOnlyList<string>? schemaIds)
        {
            if (schemaIds == null || schemaIds.Count == 0 || shard.SchemaIds.Count == 0)
                return true;

            var requestedSchemaIds = schemaIds.ToHashSet(StringComparer.Ordinal);
            if (shard.SchemaIds.Any(requestedSchemaIds.Contains))
                return true;

            foreach (var shardSchemaId in shard.SchemaIds)
            {
                try
                {
                    if (CultNetSchemaAliasMatching.MatchesAny(schemaIds, _cache.Registry.GetRequiredBySchemaId(shardSchemaId)))
                        return true;
                }
                catch (InvalidOperationException)
                {
                    // Shards can advertise schemas that are not loaded in this process.
                }
            }

            return false;
        }

        internal static CultNetShardDescriptorMessage ToMessage(CultNetShardDescriptor shard)
        {
            return new CultNetShardDescriptorMessage
            {
                ShardId = shard.ShardId,
                OwnerRuntimeId = shard.OwnerRuntimeId,
                Epoch = shard.Epoch,
                IsPrimary = shard.IsPrimary,
                SchemaIds = shard.SchemaIds.ToArray(),
                KeyPrefix = shard.KeyPrefix,
                PrimaryEndpoints = shard.PrimaryEndpoints.ToArray(),
                ReplicaEndpoints = shard.ReplicaEndpoints.ToArray(),
                ReadReplicaEndpoints = shard.ReadReplicaEndpoints.ToArray(),
                Region = shard.Region
            };
        }

        private void ThrowIfDisposed()
        {
            if (_disposed)
            {
                throw new ObjectDisposedException(nameof(CultNetDatabase));
            }
        }
    }
}