CultLib/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

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

namespace GameCult.Networking
{
    /// <summary>Controls whether subscription payloads become replica state or remain ephemeral typed events.</summary>
    public enum CultNetDatabaseSubscriptionDeliveryMode
    {
        /// <summary>Decode each payload and write it into the local CultCache before notifying observers.</summary>
        ReplicateToCache,
        /// <summary>Decode each payload and notify observers without writing replica state.</summary>
        Live
    }

    /// <summary>
    /// Replicates selected remote database records into a local typed CultCache over one retained session.
    /// </summary>
    public sealed class CultNetDatabaseSubscriptionClient : IDisposable
    {
        private readonly ICultNetSchemaClient _client;
        private readonly CultCache _cache;
        private readonly CultNetDocumentRegistry _documents;
        private readonly ConcurrentDictionary<string, PendingSubscription> _initialSnapshots = new();
        private readonly ConcurrentDictionary<string, CultNetDatabaseSubscriptionDeliveryMode> _deliveryModes =
            new(StringComparer.Ordinal);
        // Changes of a commit that has not fully arrived, by subscription and commit id.
        // Guarded by its own lock: the queue fills it, Unsubscribe and Dispose evict. A commit that does not complete is bounded
        // by MaxPendingCommits and by MaxBufferedChanges, and a violation faults the client.
        private const int MaxPendingCommits = 64;
        // Policy: the most changes the incomplete commits may announce between them, the bound on what the buffer can hold. It
        // sits far above any commit a source makes; a commit that announces more than this could not be held at all.
        private const int MaxBufferedChanges = 1_000_000;
        private readonly Dictionary<(string SubscriptionId, string CommitId), PendingCommit> _commits = new();
        private readonly object _queueLock = new();
        private Task _queue = Task.CompletedTask;
        private bool _disposed;

        public CultNetDatabaseSubscriptionClient(
            ICultNetSchemaClient client,
            CultCache cache,
            CultNetDocumentRegistry documents)
        {
            _client = client ?? throw new ArgumentNullException(nameof(client));
            _cache = cache ?? throw new ArgumentNullException(nameof(cache));
            _documents = documents ?? throw new ArgumentNullException(nameof(documents));
            _client.OnCultNet<CultNetSnapshotResponseRawMessage>(HandleSnapshot);
            _client.OnCultNet<CultNetDatabaseChangeRawMessage>(HandleChange);
        }

        /// <summary>Raised after a live change has been applied to the local typed cache.</summary>
        public event Action<CultNetReplicatedDocumentChange>? Changed;

        /// <summary>
        /// Raised when the local cache refuses a replicated commit. The replica then differs from its source; the client applies
        /// no further changes and reports <see cref="Fault"/> until it is replaced by a new subscription.
        /// </summary>
        public event Action<Exception>? Faulted;

        /// <summary>The refusal that stopped replication, or null while the replica is intact.</summary>
        public Exception? Fault { get; private set; }

        /// <summary>Gets the local reactive cache receiving replicated documents.</summary>
        public CultCache Cache => _cache;

        /// <summary>Completes if the retained transport pump terminates with an unrecoverable error.</summary>
        public Task<Exception>? BackgroundFailure =>
            (_client as RudpCultNetSchemaClient)?.BackgroundFailure;

        /// <summary>
        /// Subscribes to one exact typed record as an ephemeral reactive value. The returned value
        /// owns filtering and unsubscribe; no received payload is written into the local cache.
        /// </summary>
        public async Task<CultNetLiveValue<TDocument>> SubscribeLiveValueAsync<TDocument>(
            string subscriptionId,
            string recordKey,
            CancellationToken cancellationToken = default)
            where TDocument : class
        {
            if (string.IsNullOrWhiteSpace(recordKey))
                throw new ArgumentException("Record key is required.", nameof(recordKey));
            var schemaId = _documents.GetByDocumentType(typeof(TDocument))?.SchemaId ??
                _cache.Registry.GetRequired<TDocument>().SchemaId;
            var value = new CultNetLiveValue<TDocument>(this, subscriptionId, recordKey);
            try
            {
                var initial = await SubscribeAsync(
                        subscriptionId,
                        recordKeys: new[] { recordKey },
                        schemaIds: new[] { schemaId },
                        deliveryMode: CultNetDatabaseSubscriptionDeliveryMode.Live,
                        cancellationToken: cancellationToken)
                    .ConfigureAwait(false);
                var documents = initial.OfType<TDocument>().ToArray();
                if (documents.Length > 1)
                    throw new InvalidOperationException(
                        $"Live value subscription '{subscriptionId}' expected at most one '{schemaId}' record '{recordKey}' but received {documents.Length}.");
                if (documents.Length == 1)
                    value.Initialize(documents[0]);
                return value;
            }
            catch
            {
                value.Dispose();
                throw;
            }
        }

        /// <summary>
        /// Starts a remote subscription and waits until its initial matching snapshot is in the local cache.
        /// </summary>
        public Task<IReadOnlyList<object>> SubscribeAsync(
            string subscriptionId,
            IEnumerable<string>? recordKeys = null,
            IEnumerable<string>? schemaIds = null,
            bool includeSnapshot = true,
            CancellationToken cancellationToken = default,
            string? consumerRuntimeId = null,
            IEnumerable<string>? bodyIds = null,
            IEnumerable<string>? supportedBodyTransports = null,
            CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache)
        {
            if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient));
            if (Fault != null)
                throw new InvalidOperationException("The replica refused a commit and differs from its source; subscribe with a new client.", Fault);
            if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing.");
            if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId));

            var messageId = Guid.NewGuid().ToString("N");
            var completion = new TaskCompletionSource<IReadOnlyList<object>>(TaskCreationOptions.RunContinuationsAsynchronously);
            if (!_initialSnapshots.TryAdd(messageId, new PendingSubscription(subscriptionId, deliveryMode, completion)))
                throw new InvalidOperationException($"Duplicate database subscription message id '{messageId}'.");
            _deliveryModes[subscriptionId] = deliveryMode;

            CancellationTokenRegistration cancellation = default;
            if (cancellationToken.CanBeCanceled)
            {
                cancellation = cancellationToken.Register(() =>
                {
                    if (_initialSnapshots.TryRemove(messageId, out var pending))
                        pending.Completion.TrySetCanceled(cancellationToken);
                });
                _ = completion.Task.ContinueWith(
                    _ => cancellation.Dispose(),
                    CancellationToken.None,
                    TaskContinuationOptions.ExecuteSynchronously,
                    TaskScheduler.Default);
            }

            try
            {
                _client.SendCultNet(new CultNetDatabaseSubscribeMessage
                {
                    MessageId = messageId,
                    SubscriptionId = subscriptionId,
                    RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
                    SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
                    IncludeSnapshot = includeSnapshot,
                    ConsumerRuntimeId = consumerRuntimeId,
                    BodyIds = bodyIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
                    SupportedBodyTransports = supportedBodyTransports?
                        .Where(value => !string.IsNullOrWhiteSpace(value))
                        .Distinct(StringComparer.OrdinalIgnoreCase)
                        .ToArray()
                });
            }
            catch
            {
                _deliveryModes.TryRemove(subscriptionId, out _);
                if (_initialSnapshots.TryRemove(messageId, out var pending))
                    pending.Completion.TrySetException(new InvalidOperationException("CultNet subscription request could not be sent."));
                throw;
            }
            return completion.Task;
        }

        /// <summary>Stops one live remote subscription.</summary>
        public void Unsubscribe(string subscriptionId)
        {
            if (_disposed) return;
            if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId));
            _deliveryModes.TryRemove(subscriptionId, out _);
            lock (_commits)
                foreach (var held in _commits.Keys.Where(key => key.SubscriptionId == subscriptionId).ToArray())
                    _commits.Remove(held);
            _client.SendCultNet(new CultNetDatabaseUnsubscribeMessage
            {
                MessageId = Guid.NewGuid().ToString("N"),
                SubscriptionId = subscriptionId
            });
        }

        public void Dispose()
        {
            if (_disposed) return;
            _disposed = true;
            foreach (var pending in _initialSnapshots.Values)
                pending.Completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)));
            _initialSnapshots.Clear();
            _deliveryModes.Clear();
            lock (_commits)
                _commits.Clear();
            _client.Dispose();
        }

        private void HandleSnapshot(CultNetSnapshotResponseRawMessage message)
        {
            Enqueue(async () =>
            {
                if (!_initialSnapshots.TryRemove(message.MessageId, out var pending)) return;
                try
                {
                    IReadOnlyList<object> applied;
                    if (pending.DeliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live)
                    {
                        applied = message.Documents.Select(_documents.DeserializeRawDocument).ToArray();
                    }
                    else
                    {
                        applied = await _documents.ApplyRawSnapshotResponseAsync(_cache, message).ConfigureAwait(false);
                    }
                    pending.Completion.TrySetResult(applied);
                }
                catch (Exception error)
                {
                    pending.Completion.TrySetException(error);
                }
            });
        }

        private void HandleChange(CultNetDatabaseChangeRawMessage message)
        {
            Enqueue(() =>
            {
                var deliveryMode = _deliveryModes.TryGetValue(message.SubscriptionId, out var configured)
                    ? configured
                    : CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache;
                // Live delivery applies nothing, so there is no commit to judge; each change is reported as it arrives.
                if (deliveryMode != CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache || message.CommitId == null || !(message.CommitCount > 1))
                    return Apply(deliveryMode, new[] { message });

                // The commit's other changes follow on this connection; the replica applies them as one commit.
                var held = (message.SubscriptionId, message.CommitId);
                PendingCommit? pending;
                lock (_commits)
                {
                    // A straggler of a subscription that was ended, or never made, is not buffered. Judged under the lock that
                    // Unsubscribe evicts under: a check before it could pass, lose the race to the eviction, and then buffer.
                    if (Fault != null || !_deliveryModes.ContainsKey(message.SubscriptionId))
                        return Task.CompletedTask;
                    var count = message.CommitCount!.Value;
                    var refusal = !_commits.TryGetValue(held, out pending)
                            ? (_commits.Count >= MaxPendingCommits
                                ? $"more than {MaxPendingCommits} commits are incomplete at once"
                                : (long)count + _commits.Sum(entry => (long)entry.Value.Announced) > MaxBufferedChanges
                                    ? $"commit '{message.CommitId}' announces {count} changes; the incomplete commits may announce {MaxBufferedChanges} between them"
                                    : null)
                            : pending.Announced != count
                                ? $"commit '{message.CommitId}' announces {pending.Announced} and {count} changes"
                                : pending.Ids.Contains(message.MessageId)
                                    ? $"commit '{message.CommitId}' delivered message '{message.MessageId}' twice"
                                    : null;
                    if (refusal != null)
                    {
                        Break(new InvalidOperationException($"CultNet subscription '{message.SubscriptionId}': {refusal}."));
                        return Task.CompletedTask;
                    }

                    if (!_commits.TryGetValue(held, out pending))
                        _commits[held] = pending = new PendingCommit(count);
                    pending.Ids.Add(message.MessageId);
                    pending.Messages.Add(message);
                    if (pending.Messages.Count < count)
                        return Task.CompletedTask;
                    _commits.Remove(held);
                }

                return Apply(deliveryMode, pending.Messages);
            });
        }

        private sealed class PendingCommit
        {
            public PendingCommit(int announced) => Announced = announced;
            public int Announced { get; }
            public List<CultNetDatabaseChangeRawMessage> Messages { get; } = new();
            public HashSet<string> Ids { get; } = new(StringComparer.Ordinal);
        }

        // The replica no longer matches its source: say so once, drop what was waiting, and apply nothing more.
        private void Break(Exception error)
        {
            lock (_commits)
            {
                if (Fault != null)
                    return;
                Fault = error;
                _commits.Clear();
            }

            Faulted?.Invoke(error);
        }

        private Task Apply(CultNetDatabaseSubscriptionDeliveryMode deliveryMode, IReadOnlyList<CultNetDatabaseChangeRawMessage> changes)
        {
            // A replica that refused a commit no longer matches its source. Applying later commits on top would hide that, so
            // nothing more is applied: the fault is surfaced and the replica is rebuilt by subscribing again.
            if (Fault != null)
                return Task.CompletedTask;
            IReadOnlyList<object?> applied;
            try
            {
                applied = deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live
                    ? changes.Select(change => change.Document == null ? null : _documents.DeserializeRawDocument(change.Document)).ToArray()
                    : _documents.ApplyRawChangesAsOneCommit(_cache, changes);
            }
            catch (Exception error)
            {
                Break(error);
                return Task.CompletedTask;
            }

            for (var index = 0; index < changes.Count; index++)
            {
                var change = changes[index];
                var removed = string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal);
                if (!removed && change.Document == null)
                    continue;
                Changed?.Invoke(new CultNetReplicatedDocumentChange(
                    change.SubscriptionId,
                    change.ChangeKind,
                    removed ? change.RecordKey ?? string.Empty : change.Document!.RecordKey,
                    removed ? change.SchemaId ?? string.Empty : change.Document!.SchemaId,
                    removed ? null : applied[index]));
            }

            return Task.CompletedTask;
        }

        private void Enqueue(Func<Task> operation)
        {
            lock (_queueLock)
            {
                _queue = _queue.ContinueWith(
                    _ => operation(),
                    TaskScheduler.Default).Unwrap();
            }
        }

        private sealed class PendingSubscription
        {
            public PendingSubscription(
                string subscriptionId,
                CultNetDatabaseSubscriptionDeliveryMode deliveryMode,
                TaskCompletionSource<IReadOnlyList<object>> completion)
            {
                SubscriptionId = subscriptionId;
                DeliveryMode = deliveryMode;
                Completion = completion;
            }

            public string SubscriptionId { get; }
            public CultNetDatabaseSubscriptionDeliveryMode DeliveryMode { get; }
            public TaskCompletionSource<IReadOnlyList<object>> Completion { get; }
        }
    }

    /// <summary>One exact typed remote record kept as an ephemeral current value.</summary>
    public sealed class CultNetLiveValue<TDocument> : IDisposable where TDocument : class
    {
        private readonly CultNetDatabaseSubscriptionClient _owner;
        private readonly string _subscriptionId;
        private readonly string _recordKey;
        private readonly object _gate = new();
        private TDocument? _current;
        private bool _hasValue;
        private bool _disposed;

        internal CultNetLiveValue(
            CultNetDatabaseSubscriptionClient owner,
            string subscriptionId,
            string recordKey)
        {
            _owner = owner ?? throw new ArgumentNullException(nameof(owner));
            _subscriptionId = string.IsNullOrWhiteSpace(subscriptionId)
                ? throw new ArgumentException("Subscription id is required.", nameof(subscriptionId))
                : subscriptionId;
            _recordKey = recordKey;
            _owner.Changed += OnChanged;
        }

        /// <summary>Raised when the provider publishes a new typed value.</summary>
        public event Action<TDocument>? Changed;

        /// <summary>Raised when the provider removes the selected record.</summary>
        public event Action? Removed;

        /// <summary>Gets whether the selected record currently exists.</summary>
        public bool HasValue
        {
            get { lock (_gate) return _hasValue; }
        }

        /// <summary>Gets the current value, or fails when the record has not been published.</summary>
        public TDocument Current
        {
            get
            {
                lock (_gate)
                    return _hasValue
                        ? _current!
                        : throw new InvalidOperationException(
                            $"Live value '{_recordKey}' is not currently published.");
            }
        }

        internal void Initialize(TDocument document)
        {
            lock (_gate)
            {
                if (_disposed || _hasValue) return;
                _current = document;
                _hasValue = true;
            }
        }

        public void Dispose()
        {
            lock (_gate)
            {
                if (_disposed) return;
                _disposed = true;
            }
            _owner.Changed -= OnChanged;
            _owner.Unsubscribe(_subscriptionId);
        }

        private void OnChanged(CultNetReplicatedDocumentChange change)
        {
            if (!string.Equals(change.SubscriptionId, _subscriptionId, StringComparison.Ordinal) ||
                !string.Equals(change.RecordKey, _recordKey, StringComparison.Ordinal))
                return;

            if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal))
            {
                lock (_gate)
                {
                    if (_disposed) return;
                    _current = null;
                    _hasValue = false;
                }
                Removed?.Invoke();
                return;
            }

            if (change.Document is not TDocument document)
                return;
            lock (_gate)
            {
                if (_disposed) return;
                _current = document;
                _hasValue = true;
            }
            Changed?.Invoke(document);
        }
    }

    /// <summary>Describes one typed document change after local replication.</summary>
    public sealed class CultNetReplicatedDocumentChange
    {
        public CultNetReplicatedDocumentChange(
            string subscriptionId,
            string changeKind,
            string recordKey,
            string schemaId,
            object? document)
        {
            SubscriptionId = subscriptionId;
            ChangeKind = changeKind;
            RecordKey = recordKey;
            SchemaId = schemaId;
            Document = document;
        }

        public string SubscriptionId { get; }
        public string ChangeKind { get; }
        public string RecordKey { get; }
        public string SchemaId { get; }
        public object? Document { get; }
    }
}