using System; using System.IO; using System.Threading.Tasks; using Aetheria.State.Documents; using GameCult.Aetheria.State.Verse; using GameCult.Caching; using GameCult.Caching.MessagePack; using GameCult.Mesh; using GameCult.Networking; using R3; namespace Aetheria.State; public sealed class AetheriaStateNode : IAsyncDisposable, IDisposable { private readonly CultMeshNode _node; private CultMeshDocumentHandle? _runtimeCatalog; private AetheriaStateNode(string statePath, string runtimeId, CultMeshNode node) { StatePath = statePath; RuntimeId = string.IsNullOrWhiteSpace(runtimeId) ? "aetheria-local" : runtimeId; _node = node; } public string StatePath { get; } public string RuntimeId { get; } public CultMeshNode MeshNode => _node; public CultCache Cache => _node.Cache; public CultNetDatabase Database => _node.Database; public static async Task OpenAsync( string statePath, string runtimeId = "aetheria-local", bool startServer = false, bool enableDurableShardLogs = true) { if (string.IsNullOrWhiteSpace(statePath)) { throw new ArgumentException("State path must be non-empty.", nameof(statePath)); } Directory.CreateDirectory(Path.GetDirectoryName(Path.GetFullPath(statePath)) ?? "."); var cacheRegistry = AetheriaDocumentRegistry.CreateCultCacheRegistry(); var node = await CultMesh.CreateNodeAsync( statePath, new CultMeshNodeOptions { StartServer = startServer, EnableDurableShardLogs = enableDurableShardLogs, CacheOptions = new CultCacheOpenOptions { Registry = cacheRegistry, PullOnOpen = true, StoreFlushOnDispose = true, UseDirectoryStore = true }, DatabaseOptions = new CultNetDatabaseOptions { RuntimeId = runtimeId, DocumentRegistry = AetheriaDocumentRegistry.CreateCultNetRegistry(cacheRegistry) } }).ConfigureAwait(false); return new AetheriaStateNode(statePath, runtimeId, node); } public CultMeshDocumentHandle Document(CultRecordKey key) where TDocument : class { return CultMesh.Document( key.ToString(), CultMesh.Verse("aetheria.local", RuntimeId), async _ => (await Database.GetAsync(key).ConfigureAwait(false))!, _ => Database.WatchRecord(key) .Where(change => change.Document != null) .Select(change => change.Document!), sources: new[] { CultMesh.ProjectionSource(key.ToString()) }, routeHint: new CultMeshRouteHint(CultMeshLocalityKind.SharedMemory, "Aetheria typed state node document")); } public CultMeshDocumentHandle RuntimeCatalog() { return _runtimeCatalog ??= CultMesh.Document( "aetheria.catalog.runtime", CultMesh.Verse("aetheria.local", RuntimeId), _ => Task.FromResult(AetheriaRuntimeCatalogStore.OpenReadOnly(StatePath)), _ => Database.WatchRecord( AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest) .Where(change => change.Document != null) .Select(_ => AetheriaRuntimeCatalogStore.OpenReadOnly(StatePath)), sources: new[] { CultMesh.ProjectionSource("catalog:aetheria.runtime"), CultMesh.ProjectionSource( "aetheria.catalog.runtime", AetheriaRuntimeCatalogSnapshot.SchemaId, "managed Aetheria runtime catalog document") }, routeHint: new CultMeshRouteHint(CultMeshLocalityKind.SharedMemory, "Aetheria typed catalog state")); } public CultMeshMutableStatePointer ProviderAdvertisement() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.DaemonProviderAdvertisement); } public CultMeshMutableStatePointer Health() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.DaemonHealth); } public CultMeshMutableStatePointer CommandBoundary() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.DaemonCommandBoundary); } public CultMeshMutableStatePointer VerseAuthorityPolicy() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.VerseAuthorityPolicy); } public CultMeshMutableStatePointer LatestFrame() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest); } public CultMeshMutableStatePointer LatestSoaView() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.DaemonSoaViewLatest); } public CultMeshMutableStatePointer StarbridgeScenario() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.StarbridgeScenarioLatest); } public CultMeshMutableStatePointer StarbridgeSession() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.StarbridgeSessionLatest); } public CultMeshMutableStatePointer StarbridgeSessionSummary() { return MutableDocumentPointer( AetheriaRuntimeVerseRecordKeys.StarbridgeSessionSummary); } public CultMeshMutableStatePointer DaemonGameSurface() { return MutableDocumentPointer(AetheriaRuntimeVerseRecordKeys.DaemonGameSurface); } public CultMeshMutableStatePointer DaemonGameTuiSurface() { return MutableDocumentPointer(AetheriaRuntimeVerseRecordKeys.DaemonGameTuiSurface); } public CultMeshMutableStatePointer DaemonEditorSurface() { return MutableDocumentPointer(AetheriaRuntimeVerseRecordKeys.DaemonEditorSurface); } public CultMeshMutableStatePointer DaemonEditorTuiSurface() { return MutableDocumentPointer(AetheriaRuntimeVerseRecordKeys.DaemonEditorTuiSurface); } public CultMeshMutableStatePointer CatalogSurface() { return MutableDocumentPointer(new CultRecordKey(AetheriaCatalogSurfaceProjector.SurfaceKey)); } public CultMeshMutableStatePointer OperationsSurface() { return MutableDocumentPointer(new CultRecordKey(AetheriaOperationsSurfaceProjector.SurfaceKey)); } public CultMeshMutableStatePointer PlayerSettingsSurface() { return MutableDocumentPointer(new CultRecordKey(AetheriaPlayerSettingsSurfaceProjector.SurfaceKey)); } public CultMeshMutableStatePointer ProviderAdvertisementSurface() { return MutableDocumentPointer( new CultRecordKey(AetheriaProviderAdvertisementProjector.AdvertisementKey)); } public CultMeshMutableStatePointer RuntimeSession(string runtimeId) { return MutableDocumentPointer(RuntimeSessionKey(runtimeId)); } public CultMeshMutableStatePointer PlayerSettings() { return MutableDocumentPointer(PlayerSettingsKey); } public CultMeshMutableStatePointer VerseHostSettings() { return MutableDocumentPointer(VerseHostSettingsKey); } public CultMeshMutableStatePointer EveCommandAcceptanceStatus() { return MutableDocumentPointer(EveCommandAcceptanceStatusKey); } public CultMeshMutableStatePointer World() { return MutableDocumentPointer(new CultRecordKey("global:aetheria.world_state.v1")); } public CultMeshMutableStatePointer MigrationLedger() { return MutableDocumentPointer( new CultRecordKey("global:aetheria.migration_ledger.v1")); } public CultMeshMutableStatePointer LegacyCatalogQuarantine() { return MutableDocumentPointer( new CultRecordKey("global:aetheria.legacy_catalog_quarantine.v1")); } public CultMeshMutableStatePointer ItemDefinition(CultRecordKey key) { return MutableDocumentPointer(key); } public CultMeshMutableStatePointer ItemDefinitionByLegacyId(string legacyId) { return ItemDefinition(AetheriaCatalogKeys.ItemDefinitionFromLegacyId(legacyId)); } public CultMeshMutableStatePointer Corporation(CultRecordKey key) { return MutableDocumentPointer(key); } public CultMeshMutableStatePointer CorporationByLegacyId(string legacyId) { return Corporation(AetheriaCatalogKeys.CorporationFromLegacyId(legacyId)); } public CultMeshMutableStatePointer NameFile(CultRecordKey key) { return MutableDocumentPointer(key); } public CultMeshMutableStatePointer NameFileByLegacyId(string legacyId) { return NameFile(AetheriaCatalogKeys.NameFileFromLegacyId(legacyId)); } public CultMeshMutableStatePointer TradeValuePolicy() { return MutableDocumentPointer( new CultRecordKey(AetheriaTradeValuePolicy.RecordKey)); } public CultMeshMutableStatePointer LoadoutTemplate(CultRecordKey key) { return MutableDocumentPointer(key); } public CultMeshMutableStatePointer RunState(CultRecordKey key) { return MutableDocumentPointer(key); } public CultMeshMutableStatePointer ZoneState(CultRecordKey key) { return MutableDocumentPointer(key); } public CultMeshMutableStatePointer EntitySnapshot(CultRecordKey key) { return MutableDocumentPointer(key); } public Task> SubmitDaemonCommandAsync( AetheriaRuntimeDaemonCommandDocument command) { if (command == null) throw new ArgumentNullException(nameof(command)); command.Schema = AetheriaRuntimeDaemonSchemas.Command; if (string.IsNullOrWhiteSpace(command.CommandId)) command.CommandId = Guid.NewGuid().ToString("N"); if (string.IsNullOrWhiteSpace(command.IssuedAtUtc)) command.IssuedAtUtc = DateTime.UtcNow.ToString("O"); return Database.PutAsync(DaemonCommandKey(command.CommandId), command); } public IReadOnlyList ReadObservedDaemonCommands() { return Cache .GetAll() .OrderBy(command => command.IssuedAtUtc ?? "", StringComparer.Ordinal) .ThenBy(command => command.CommandId ?? "", StringComparer.Ordinal) .ToArray(); } public Task> PutCommittedCommandFactAsync( AetheriaRuntimeCommittedCommandFactDocument fact) { if (fact == null) throw new ArgumentNullException(nameof(fact)); fact.Schema = AetheriaRuntimeDaemonSchemas.CommittedCommandFact; if (string.IsNullOrWhiteSpace(fact.FactId)) fact.FactId = Guid.NewGuid().ToString("N"); if (string.IsNullOrWhiteSpace(fact.CommittedAtUtc)) fact.CommittedAtUtc = DateTime.UtcNow.ToString("O"); return Database.PutAsync( new CultRecordKey(AetheriaRuntimeCommittedCommandFactDocument.CreateRecordKey(fact.FactId)), fact); } public IReadOnlyList ReadCommittedCommandFacts() { return Cache .GetAll() .OrderBy(fact => fact.CommittedAtUtc ?? "", StringComparer.Ordinal) .ThenBy(fact => fact.FactId ?? "", StringComparer.Ordinal) .ToArray(); } public IReadOnlyList ReadAuthorityLeases() { return Cache.GetAll().ToArray(); } private static CultRecordKey DaemonCommandKey(string commandId) { return new CultRecordKey($"daemon:commands:{StableToken(commandId)}:gamecult.aetheria.daemon_command.v1"); } public Task> SubmitEveCommandAsync( AetheriaRuntimeEveCommandDocument command) { if (command == null) throw new ArgumentNullException(nameof(command)); AetheriaRuntimeEveCommandClient.NormalizeDocument(command); command.Schema = AetheriaRuntimeEveCommandDocument.SchemaId; if (string.IsNullOrWhiteSpace(command.CommandId)) command.CommandId = Guid.NewGuid().ToString("N"); if (string.IsNullOrWhiteSpace(command.IssuedAtUtc)) command.IssuedAtUtc = DateTime.UtcNow.ToString("O"); return Database.PutAsync(EveCommandKey(command.CommandId), command); } public IReadOnlyList ReadObservedEveCommands() { return Cache .GetAll() .Select(AetheriaRuntimeEveCommandClient.NormalizeDocument) .OrderBy(command => command.IssuedAtUtc ?? "", StringComparer.Ordinal) .ThenBy(command => command.CommandId ?? "", StringComparer.Ordinal) .ToArray(); } private static CultRecordKey EveCommandKey(string commandId) { return new CultRecordKey($"eve:commands:{StableToken(commandId)}:gamecult.eve.command.v1"); } public Task FlushAsync(bool soft = false) { return _node.FlushAsync(soft); } private static string StableToken(string value) { if (string.IsNullOrWhiteSpace(value)) return "empty"; var chars = value .Select(character => char.IsLetterOrDigit(character) ? character : '-') .ToArray(); var token = new string(chars).Trim('-').ToLowerInvariant(); while (token.Contains("--", StringComparison.Ordinal)) token = token.Replace("--", "-", StringComparison.Ordinal); return string.IsNullOrWhiteSpace(token) ? "empty" : token; } private static CultRecordKey RuntimeSessionKey(string runtimeId) { return new CultRecordKey($"runtime:{runtimeId}:aetheria.runtime_session.v1"); } private static CultRecordKey PlayerSettingsKey { get; } = new("global:aetheria.player_settings.v1"); private static CultRecordKey VerseHostSettingsKey { get; } = new("global:aetheria.verse_host_settings.v1"); private static CultRecordKey EveCommandAcceptanceStatusKey { get; } = new("global:aetheria.eve_command_acceptance_status.v1"); private CultMeshMutableStatePointer MutableDocumentPointer(CultRecordKey key) where T : class { return CultMesh.MutableStatePointer( key.ToString(), _ => Database.GetAsync(key), _ => Database.WatchRecord(key) .Where(change => change.Document != null) .Select(change => change.Document!), async (_, value) => { await Database.PutAsync(key, value).ConfigureAwait(false); }, sources: [ CultMesh.ProjectionSource(key.ToString()) ]); } public void Dispose() { _node.Dispose(); } public async ValueTask DisposeAsync() { await FlushAsync().ConfigureAwait(false); Dispose(); } }