using System; using System.Globalization; using System.IO; using System.Net; using System.Net.Http; using System.Threading; using System.Threading.Tasks; using game.Account; namespace game.PlayerDataStorage { public enum PlayerKVSessionOpenError { None, Validation, Unauthenticated, Busy, RecoveryRequired, StorageError, ReadFailed, Unauthorized, Stale, Cancelled } public sealed class PlayerKVSessionOpenResult { public bool Success => Error == PlayerKVSessionOpenError.None; public PlayerKVSessionOpenError Error { get; } public PlayerKVError TransportError { get; } public PlayerKVSessionOwner Owner { get; } internal PlayerKVSessionOpenResult(PlayerKVSessionOpenError error, PlayerKVError transportError = PlayerKVError.None, PlayerKVSessionOwner owner = null) { Error = error; TransportError = transportError; Owner = owner; } } /// Owns one immutable account/epoch/cache/transport tuple. public sealed class PlayerKVSessionOwner : IDisposable { sealed class OperationLease : IDisposable { PlayerKVSessionOwner owner; internal OperationLease(PlayerKVSessionOwner owner) { this.owner = owner; } public void Dispose() { Interlocked.Exchange(ref owner, null)?.ExitOperation(); } } readonly object admission = new object(); readonly S101AccountSession session; readonly CookieContainer cookies; readonly CancellationTokenSource lifetime; readonly Func isCurrent; readonly PlayerKVSyncState state; readonly PlayerKVSyncCoordinator coordinator; readonly TaskCompletionSource drained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); readonly string ownerId = Guid.NewGuid().ToString("N"); int activeOperations; bool closing, closed; TaskCompletionSource closeCompletion; public string OwnerInstanceId => ownerId; public Uri ApiBaseUri { get; } public string PlayerId { get; } public long Epoch { get; } public PlayerKVSyncStatus Status => state.Status; PlayerKVSessionOwner(S101AccountSession session, CookieContainer cookies, Uri api, string playerId, long epoch, Func isCurrent, CancellationTokenSource lifetime, PlayerKVSyncState state, HttpClient http) { this.session = session; this.cookies = cookies; ApiBaseUri = api; PlayerId = playerId; Epoch = epoch; this.isCurrent = isCurrent; this.lifetime = lifetime; this.state = state; coordinator = new PlayerKVSyncCoordinator(state, new PlayerKVClient(http, api), http, ownerId, epoch, IsCapturedOwnerCurrent, lifetime.Token); } public static Task CreateAsync(S101AccountSession session, string cacheRoot, string expectedPlayerId, Uri expectedApiBaseUri, long epoch, Func isCurrent, CancellationToken lifetimeToken = default) => CreateAsync(session, cacheRoot, expectedPlayerId, expectedApiBaseUri, epoch, isCurrent, lifetimeToken, jar => new HttpClientHandler { CookieContainer = jar, UseCookies = true, AllowAutoRedirect = false }); internal static async Task CreateAsync(S101AccountSession session, string cacheRoot, string expectedPlayerId, Uri expectedApiBaseUri, long epoch, Func isCurrent, CancellationToken lifetimeToken, Func handlerFactory, TimeSpan? transportTimeout = null) { if (session == null || expectedApiBaseUri == null || isCurrent == null || handlerFactory == null || epoch < 0 || !long.TryParse(expectedPlayerId, NumberStyles.None, CultureInfo.InvariantCulture, out var numericId) || numericId <= 0) return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Validation); var playerId = numericId.ToString(CultureInfo.InvariantCulture); Uri expectedApi; try { expectedApi = NormalizeApi(expectedApiBaseUri); } catch (ArgumentException) { return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Validation); } if (lifetimeToken.IsCancellationRequested) return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Cancelled, PlayerKVError.Cancelled); if (!session.IsAuthenticated || session.Cookies == null) return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Unauthenticated); var capturedCookies = session.Cookies; var capturedApi = NormalizeApi(session.ApiBaseUri); if (session.PlayerId != playerId || capturedApi != expectedApi || !isCurrent(epoch)) return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Stale); var opened = PlayerKVSyncState.Open(cacheRoot, capturedApi, playerId); if (!opened.Success) return new PlayerKVSessionOpenResult(MapOpen(opened.Error)); HttpClient http = null; CancellationTokenSource lifetime = null; PlayerKVSessionOwner owner = null; try { var handler = handlerFactory(capturedCookies); if (handler == null) throw new ArgumentException("Transport factory returned null."); http = new HttpClient(handler) { Timeout = transportTimeout ?? TimeSpan.FromSeconds(15) }; lifetime = CancellationTokenSource.CreateLinkedTokenSource(lifetimeToken); owner = new PlayerKVSessionOwner(session, capturedCookies, capturedApi, playerId, epoch, isCurrent, lifetime, opened.State, http); http = null; lifetime = null; PlayerKVResult read; using (var lease = owner.TryEnter()) { if (lease == null) return await owner.FailCreate(PlayerKVSessionOpenError.Cancelled); read = await owner.coordinator.InitializeAsync(); } if (!read.Success) return await owner.FailCreate(read.Error == PlayerKVError.Unauthorized ? PlayerKVSessionOpenError.Unauthorized : read.Error == PlayerKVError.Cancelled ? PlayerKVSessionOpenError.Cancelled : PlayerKVSessionOpenError.ReadFailed, read.Error); if (!owner.IsCapturedOwnerCurrent()) return await owner.FailCreate(PlayerKVSessionOpenError.Stale, PlayerKVError.Cancelled); return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.None, owner: owner); } catch (OperationCanceledException) { if (owner != null) return await owner.FailCreate(PlayerKVSessionOpenError.Cancelled, PlayerKVError.Cancelled); opened.State.Dispose(); return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Cancelled, PlayerKVError.Cancelled); } catch (Exception) { if (owner != null) return await owner.FailCreate(PlayerKVSessionOpenError.Validation, PlayerKVError.Protocol); http?.Dispose(); lifetime?.Dispose(); opened.State.Dispose(); return new PlayerKVSessionOpenResult(PlayerKVSessionOpenError.Validation, PlayerKVError.Protocol); } } /// Reads the existing state under an operation lease; does not open another cache. public PlayerKVLocalReadResult ReadLocalSnapshot() { using (var lease = TryEnter()) { if (lease == null) return ReadFailure(PlayerKVLocalReadError.StaleOrClosing); try { var data = state.ReadLocal(); lock (admission) { if (!IsCapturedOwnerCurrent() || closing || closed) return ReadFailure(PlayerKVLocalReadError.StaleOrClosing); return new PlayerKVLocalReadResult(PlayerKVLocalReadError.None, ownerId, PlayerId, Epoch, data); } } catch (InvalidOperationException) { lock (admission) return ReadFailure(!IsCapturedOwnerCurrent() || closing || closed ? PlayerKVLocalReadError.StaleOrClosing : PlayerKVLocalReadError.NotReady); } } } PlayerKVLocalReadResult ReadFailure(PlayerKVLocalReadError error) => new PlayerKVLocalReadResult(error, ownerId, PlayerId, Epoch); public PlayerKVEditResult Update(string key, string value, bool remoteStore = true) { using (var lease = TryEnter()) return lease == null ? ClosedEdit() : coordinator.Update(key, value, remoteStore); } public PlayerKVEditResult Delete(string key) { using (var lease = TryEnter()) return lease == null ? ClosedEdit() : coordinator.Delete(key); } public Task FlushAsync(CancellationToken waitToken = default) { var lease = TryEnter(); if (lease == null) return Task.FromResult(new PlayerKVFlushResult(0, 0, 1, 0, PlayerKVError.Cancelled, state.Status)); return WaitForFlushAsync(RunFlushAsync(lease), waitToken); } /// Runs a complete recovery Get on the same serialized channel and fixed owner transport. public Task RefreshAsync(CancellationToken waitToken = default) { var lease = TryEnter(); if (lease == null) return Task.FromResult(new PlayerKVResult(PlayerKVError.Cancelled)); var shared = RunRefreshAsync(lease); return WaitForRefreshAsync(shared, waitToken); } public Task WaitForTicketAsync(PlayerKVSyncTicket ticket, CancellationToken cancellationToken = default) => coordinator.WaitForTicketAsync(ticket, cancellationToken); // Watermark capture must happen inside the lease scope: a failure there still // returns the operation lease, or CloseAsync would wait for this operation forever. async Task RunFlushAsync(OperationLease lease) { using (lease) { try { var watermark = coordinator.CaptureWatermark(out var targetSequences); return await coordinator.FlushThroughAsync(watermark, targetSequences); } catch (Exception) { return new PlayerKVFlushResult(0, 0, 1, 0, PlayerKVError.Protocol, state.Status); } } } async Task RunRefreshAsync(OperationLease lease) { using (lease) { try { return await coordinator.InitializeAsync(); } catch (OperationCanceledException) { return new PlayerKVResult(PlayerKVError.Cancelled); } catch (ObjectDisposedException) { return new PlayerKVResult(PlayerKVError.Cancelled); } catch (HttpRequestException) { return new PlayerKVResult(PlayerKVError.Network); } catch (IOException) { return new PlayerKVResult(PlayerKVError.Network); } catch (Exception) { return new PlayerKVResult(PlayerKVError.Protocol); } } } static async Task WaitForRefreshAsync(Task shared, CancellationToken token) { if (!token.CanBeCanceled) return await shared; if (token.IsCancellationRequested) return new PlayerKVResult(PlayerKVError.Cancelled); var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using (token.Register(() => cancelled.TrySetResult(true))) { if (await Task.WhenAny(shared, cancelled.Task) == shared) return await shared; } return new PlayerKVResult(PlayerKVError.Cancelled); } async Task WaitForFlushAsync(Task shared, CancellationToken token) { if (!token.CanBeCanceled) return await shared; if (token.IsCancellationRequested) return new PlayerKVFlushResult(0, 0, 1, 0, PlayerKVError.Cancelled, state.Status); var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using (token.Register(() => cancelled.TrySetResult(true))) { if (await Task.WhenAny(shared, cancelled.Task) == shared) return await shared; } return new PlayerKVFlushResult(0, 0, 1, 0, PlayerKVError.Cancelled, state.Status); } OperationLease TryEnter() { lock (admission) { if (closing || closed || !IsCapturedOwnerCurrent()) return null; activeOperations++; return new OperationLease(this); } } void ExitOperation() { lock (admission) { activeOperations--; if (closing && activeOperations == 0) drained.TrySetResult(true); } } public Task CloseAsync() { TaskCompletionSource completion; lock (admission) { if (closeCompletion != null) return closeCompletion.Task; closing = true; completion = closeCompletion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); if (activeOperations == 0) drained.TrySetResult(true); } _ = CloseCoreAsync(completion); return completion.Task; } async Task CloseCoreAsync(TaskCompletionSource completion) { try { lifetime.Cancel(); coordinator.CloseTickets(); await drained.Task; coordinator.Dispose(); state.Dispose(); lifetime.Dispose(); lock (admission) closed = true; completion.TrySetResult(true); } catch (Exception ex) { completion.TrySetException(ex); } } async Task FailCreate(PlayerKVSessionOpenError error, PlayerKVError transport = PlayerKVError.None) { await CloseAsync(); return new PlayerKVSessionOpenResult(error, transport); } bool IsCapturedOwnerCurrent() { if (lifetime.IsCancellationRequested || !isCurrent(Epoch)) return false; return session.IsAuthenticated && ReferenceEquals(session.Cookies, cookies) && session.PlayerId == PlayerId && NormalizeApi(session.ApiBaseUri) == ApiBaseUri; } static Uri NormalizeApi(Uri api) { if (api == null || !api.IsAbsoluteUri || (api.Scheme != "http" && api.Scheme != "https") || api.UserInfo.Length != 0 || api.Query.Length != 0 || api.Fragment.Length != 0) throw new ArgumentException("Invalid API base URI."); return new Uri(api.AbsoluteUri.TrimEnd('/') + "/"); } static PlayerKVSessionOpenError MapOpen(PlayerKVOpenError error) { if (error == PlayerKVOpenError.Busy) return PlayerKVSessionOpenError.Busy; if (error == PlayerKVOpenError.RecoveryRequired) return PlayerKVSessionOpenError.RecoveryRequired; if (error == PlayerKVOpenError.StorageError) return PlayerKVSessionOpenError.StorageError; return PlayerKVSessionOpenError.Validation; } PlayerKVEditResult ClosedEdit() => new PlayerKVEditResult( new PlayerKVStateResult(PlayerKVChangeError.Blocked, revision: state.Inspect().Revision)); public void Dispose() { _ = CloseAsync().ContinueWith(task => { var observed = task.Exception; }, CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); } } }