using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using GameCult.Caching; using R3; namespace GameCult.Networking { /// /// Describes the kind of domain change observed through a distributed CultCache surface. /// public enum CultNetDatabaseChangeKind { /// /// A document was added. /// Added, /// /// A document was updated. /// Updated, /// /// A document was removed. /// Removed, /// /// A document was accepted through compatible schema migration. /// SchemaMigrated, /// /// A document change was rejected. /// Rejected } /// /// Describes one shard owned or observed by a CultNet database surface. /// public sealed class CultNetShardDescriptor { /// /// Creates a shard descriptor. /// public CultNetShardDescriptor( string shardId, string ownerRuntimeId, long epoch, bool isPrimary, IEnumerable? schemaIds = null, string? keyPrefix = 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(); KeyPrefix = keyPrefix; } /// /// Gets the stable shard identifier. /// public string ShardId { get; } /// /// Gets the runtime that currently owns writes for this shard. /// public string OwnerRuntimeId { get; } /// /// Gets the monotonic shard authority epoch. /// public long Epoch { get; } /// /// Gets whether the local database instance is authoritative for writes to this shard. /// public bool IsPrimary { get; } /// /// Gets the schema ids governed by this shard. Empty means all schemas. /// public IReadOnlyList SchemaIds { get; } /// /// Gets the optional record-key prefix governed by this shard. /// public string? KeyPrefix { get; } /// /// Creates a primary shard that accepts every schema and key. /// public static CultNetShardDescriptor PrimaryAll(string ownerRuntimeId = "local", string shardId = "primary") { return new CultNetShardDescriptor(shardId, ownerRuntimeId, epoch: 1, isPrimary: true); } /// /// Creates a read-only shard descriptor. /// public static CultNetShardDescriptor ReadOnly( string shardId, string ownerRuntimeId, long epoch = 1, IEnumerable? schemaIds = null, string? keyPrefix = null) { return new CultNetShardDescriptor(shardId, ownerRuntimeId, epoch, isPrimary: false, schemaIds, keyPrefix); } internal bool Matches(string schemaId, CultRecordKey key) { var schemaMatches = SchemaIds.Count == 0 || SchemaIds.Contains(schemaId, StringComparer.Ordinal); 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; } } /// /// Options for creating a distributed CultCache database surface. /// public sealed class CultNetDatabaseOptions { /// /// Gets or sets the local runtime id advertised as shard owner for default primary shards. /// public string RuntimeId { get; set; } = "local"; /// /// Gets or sets explicit shard descriptors. When omitted, this instance owns one primary shard for all records. /// public IReadOnlyList? Shards { get; set; } /// /// Gets or sets the document registry used to create and apply raw CultNet document messages. /// public CultNetDocumentRegistry? DocumentRegistry { get; set; } } /// /// A typed domain change emitted by a CultNet database surface. /// public sealed class CultNetDatabaseChange where T : class { /// /// Creates a database change. /// 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; } /// /// Gets the change kind. /// public CultNetDatabaseChangeKind Kind { get; } /// /// Gets the changed record key. /// public CultRecordKey Key { get; } /// /// Gets the schema id of the changed document. /// public string SchemaId { get; } /// /// Gets the shard that accepted or rejected the change. /// public CultNetShardDescriptor Shard { get; } /// /// Gets the current document for added and updated events. /// public T? Document { get; } /// /// Gets the previous document for updated and removed events. /// public T? PreviousDocument { get; } /// /// Gets an optional diagnostic message. /// public string? Message { get; } } /// /// Raised when a database write targets a shard this node does not own. /// public sealed class CultNetShardAuthorityException : InvalidOperationException { /// /// Creates a shard authority exception. /// public CultNetShardAuthorityException(CultNetShardDescriptor shard, string message) : base(message) { Shard = shard; } /// /// Gets the shard that rejected the write. /// public CultNetShardDescriptor Shard { get; } } /// /// Database-style CultNet facade over a CultCache shard set. /// public sealed class CultNetDatabase : IDisposable { private readonly CultCache _cache; private readonly CultNetDocumentRegistry _documents; private readonly List _shards; private readonly Subject _changes = new(); private bool _disposed; /// /// Creates a database surface over a CultCache instance. /// public CultNetDatabase(CultCache cache, CultNetDatabaseOptions? options = null) { _cache = cache ?? throw new ArgumentNullException(nameof(cache)); options ??= new CultNetDatabaseOptions(); _documents = options.DocumentRegistry ?? new CultNetDocumentRegistry(cache.Registry); _shards = (options.Shards == null || options.Shards.Count == 0 ? new[] { CultNetShardDescriptor.PrimaryAll(options.RuntimeId) } : options.Shards) .ToList(); _cache.OnUpdate += PublishCacheUpdate; } /// /// Gets the local cache backing this database surface. /// public CultCache Cache => _cache; /// /// Gets the document registry used by the raw CultNet lane. /// public CultNetDocumentRegistry Documents => _documents; /// /// Gets the shard map known to this database surface. /// public IReadOnlyList Shards => _shards; /// /// Gets a document by key. /// public Task GetAsync(CultRecordKey key) where T : class { ThrowIfDisposed(); return Task.FromResult(_cache.Get(key)); } /// /// Adds or replaces a document at a specific key. /// public async Task> PutAsync(CultRecordKey key, T document) where T : class { ThrowIfDisposed(); if (document == null) throw new ArgumentNullException(nameof(document)); var descriptor = _cache.Registry.GetRequired(); var shard = ResolveShard(descriptor.SchemaId, key); EnsurePrimary(shard, descriptor.SchemaId, key); var previous = _cache.Get(key); var handle = await _cache.UpsertAsync(document, new CultRecordHandle(key)).ConfigureAwait(false); Publish(new CultNetDatabaseChange( previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, key, descriptor.SchemaId, shard, document, previous)); return handle; } /// /// Deletes a document by key. /// public Task DeleteAsync(CultRecordKey key) where T : class { ThrowIfDisposed(); var descriptor = _cache.Registry.GetRequired(); var shard = ResolveShard(descriptor.SchemaId, key); EnsurePrimary(shard, descriptor.SchemaId, key); var previous = _cache.Get(key); if (previous != null) { _cache.Remove(new CultRecordHandle(key)); Publish(new CultNetDatabaseChange( CultNetDatabaseChangeKind.Removed, key, descriptor.SchemaId, shard, document: null, previousDocument: previous)); } return Task.CompletedTask; } /// /// Applies a raw document mutation after checking shard authority. /// public async Task 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 shard = ResolveShard(message.Document.SchemaId, key); EnsurePrimary(shard, message.Document.SchemaId, key); var descriptor = _cache.Registry.GetRequiredBySchemaId(message.Document.SchemaId); var previous = _cache.Get(key); var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false); PublishUntyped( descriptor.DocumentType, previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, key, descriptor.SchemaId, shard, document, previous); return document; } /// /// Applies a raw document delete after checking shard authority. /// public Task ApplyDeleteAsync(CultNetDocumentDeleteMessage message) { ThrowIfDisposed(); if (message == null) throw new ArgumentNullException(nameof(message)); var key = new CultRecordKey(message.RecordKey); var shard = ResolveShard(message.SchemaId, key); EnsurePrimary(shard, message.SchemaId, key); var descriptor = _cache.Registry.GetRequiredBySchemaId(message.SchemaId); var previous = _cache.Get(key); if (previous == null) { return Task.CompletedTask; } var removeMethod = typeof(CultCache).GetMethod(nameof(CultCache.Remove))! .MakeGenericMethod(descriptor.DocumentType); var handleType = typeof(CultRecordHandle<>).MakeGenericType(descriptor.DocumentType); var handle = Activator.CreateInstance(handleType, new object[] { key }); removeMethod.Invoke(_cache, new[] { handle }); PublishUntyped( descriptor.DocumentType, CultNetDatabaseChangeKind.Removed, key, descriptor.SchemaId, shard, document: null, previousDocument: previous); return Task.CompletedTask; } /// /// Watches all changes assignable to the requested document type. /// public Observable> Watch() where T : class { ThrowIfDisposed(); return _changes .Where(change => change is CultNetDatabaseChange) .Select(change => (CultNetDatabaseChange)change); } /// /// Watches one record. /// public Observable> WatchRecord(CultRecordKey key) where T : class { return Watch().Where(change => change.Key.Equals(key)); } /// /// Watches the global record for a document type. /// public Observable> WatchGlobal() where T : class { var descriptor = _cache.Registry.GetRequired(); return WatchRecord(new CultRecordKey($"global:{descriptor.SchemaId}")); } /// /// Watches changes for a named document. /// public Observable> WatchByName(string name) where T : class { if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("Value must be non-empty.", nameof(name)); return Watch().Where(change => (change.Document != null && ReferenceEquals(_cache.GetByName(name), change.Document)) || (change.PreviousDocument != null && ReferenceEquals(_cache.GetByName(name), change.PreviousDocument))); } /// /// Watches changes for the current document mapped by an indexed value. /// public Observable> WatchByIndex(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().Where(change => (change.Document != null && ReferenceEquals(_cache.GetByIndex(alias, value), change.Document)) || (change.PreviousDocument != null && ReferenceEquals(_cache.GetByIndex(alias, value), change.PreviousDocument))); } /// /// Releases database subscriptions. /// 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 = ResolveShard(descriptor.SchemaId, 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 void Publish(CultNetDatabaseChange change) where T : class { _changes.OnNext(change); } 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 ResolveShard(string schemaId, CultRecordKey key) { return _shards.FirstOrDefault(shard => shard.Matches(schemaId, key)) ?? _shards[0]; } private static void EnsurePrimary(CultNetShardDescriptor shard, string schemaId, CultRecordKey key) { 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}'."); } private void ThrowIfDisposed() { if (_disposed) { throw new ObjectDisposedException(nameof(CultNetDatabase)); } } } }