SessionController.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
 
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
 
using Newtonsoft.Json;
 
using Serilog.Context;
 
using Tgstation.Server.Api.Models;
using Tgstation.Server.Api.Models.Internal;
using Tgstation.Server.Common.Extensions;
using Tgstation.Server.Host.Components.Chat;
using Tgstation.Server.Host.Components.Chat.Commands;
using Tgstation.Server.Host.Components.Deployment;
using Tgstation.Server.Host.Components.Engine;
using Tgstation.Server.Host.Components.Events;
using Tgstation.Server.Host.Components.Interop;
using Tgstation.Server.Host.Components.Interop.Bridge;
using Tgstation.Server.Host.Components.Interop.Topic;
using Tgstation.Server.Host.Extensions;
using Tgstation.Server.Host.System;
using Tgstation.Server.Host.Utils;
 
namespace Tgstation.Server.Host.Components.Session
{
    sealed class SessionController : Chunker, ISessionController, IBridgeHandler, IChannelSink
    {
        internal static bool LogTopicRequests { get; set; } = true;
 
        public DMApiParameters DMApiParameters => ReattachInformation;
 
        public ApiValidationStatus ApiValidationStatus
        {
            get
            {
                if (!Lifetime.IsCompleted)
                    throw new InvalidOperationException("ApiValidated cannot be checked while Lifetime is incomplete!");
                return apiValidationStatus;
            }
        }
 
        public Models.CompileJob CompileJob => ReattachInformation.Dmb.CompileJob;
 
        public EngineVersion EngineVersion => ReattachInformation.Dmb.EngineVersion;
 
        public RebootState RebootState => ReattachInformation.RebootState;
 
        public Version? DMApiVersion { get; private set; }
 
        public bool TerminationWasIntentional => terminationWasIntentional || (Lifetime.IsCompleted && Lifetime.Result == 0);
 
        public Task<LaunchResult> LaunchResult { get; }
 
        public Task<int?> Lifetime { get; }
 
        public Task OnStartup => startupTcs.Task;
 
        public Task OnReboot => rebootTcs.Task;
 
        public Task RebootGate
        {
            get => rebootGate;
            set
            {
                var tcs = new TaskCompletionSource<Task>();
                async Task Wrap()
                {
                    var toAwait = await tcs.Task;
                    await toAwait;
                    await value;
                }
 
                tcs.SetResult(Interlocked.Exchange(ref rebootGate, Wrap()));
            }
        }
 
        public Task OnPrime => primeTcs.Task;
 
        public bool DMApiAvailable => ReattachInformation.Dmb.CompileJob.DMApiVersion?.Major == DMApiConstants.InteropVersion.Major;
 
        public bool ProcessingRebootBridgeRequest => rebootBridgeRequestsProcessing > 0;
 
        public string DumpFileExtension => engineLock.UseDotnetDump
            ? ".net.dmp"
            : ".dmp";
 
        public ReattachInformation ReattachInformation { get; }
 
        public FifoSemaphore TopicSendSemaphore { get; }
 
        public long? MemoryUsage => process.MemoryUsage;
 
        public DateTimeOffset? LaunchTime => process.LaunchTime;
 
        readonly Byond.TopicSender.ITopicClient byondTopicSender;
 
        readonly IBridgeRegistration? bridgeRegistration;
 
        readonly IProcess process;
 
        readonly IEngineExecutableLock engineLock;
 
        readonly IChatTrackingContext chatTrackingContext;
 
        readonly IChatManager chat;
 
        readonly IAsyncDelayer asyncDelayer;
 
        readonly IDotnetDumpService dotnetDumpService;
 
        readonly IEventConsumer eventConsumer;
 
        readonly TaskCompletionSource initialBridgeRequestTcs;
 
        readonly Api.Models.Instance metadata;
 
        readonly CancellationTokenSource sessionDurationCts;
 
        readonly object synchronizationLock;
 
        readonly bool apiValidationSession;
 
        volatile TaskCompletionSource startupTcs;
 
        volatile TaskCompletionSource rebootTcs;
 
        volatile TaskCompletionSource primeTcs;
 
        volatile Task rebootGate;
 
        volatile Task customEventProcessingTask;
 
        volatile Task? postValidationShutdownTask;
 
        volatile uint rebootBridgeRequestsProcessing;
 
        ApiValidationStatus apiValidationStatus;
 
        bool disposed;
 
        bool released;
 
        bool terminationWasIntentional;
 
        public SessionController(
            ReattachInformation reattachInformation,
            Api.Models.Instance metadata,
            IProcess process,
            IEngineExecutableLock engineLock,
            Byond.TopicSender.ITopicClient byondTopicSender,
            IChatTrackingContext chatTrackingContext,
            IBridgeRegistrar bridgeRegistrar,
            IChatManager chat,
            IAssemblyInformationProvider assemblyInformationProvider,
            IAsyncDelayer asyncDelayer,
            IDotnetDumpService dotnetDumpService,
            IEventConsumer eventConsumer,
            ILogger<SessionController> logger,
            Func<ValueTask> postLifetimeCallback,
            uint? startupTimeout,
            bool reattached,
            bool apiValidate)
            : base(logger)
        {
            ReattachInformation = reattachInformation ?? throw new ArgumentNullException(nameof(reattachInformation));
            this.metadata = metadata ?? throw new ArgumentNullException(nameof(metadata));
            this.process = process ?? throw new ArgumentNullException(nameof(process));
            this.engineLock = engineLock ?? throw new ArgumentNullException(nameof(engineLock));
            this.byondTopicSender = byondTopicSender ?? throw new ArgumentNullException(nameof(byondTopicSender));
            this.chatTrackingContext = chatTrackingContext ?? throw new ArgumentNullException(nameof(chatTrackingContext));
            ArgumentNullException.ThrowIfNull(bridgeRegistrar);
 
            this.chat = chat ?? throw new ArgumentNullException(nameof(chat));
            ArgumentNullException.ThrowIfNull(assemblyInformationProvider);
 
            this.asyncDelayer = asyncDelayer ?? throw new ArgumentNullException(nameof(asyncDelayer));
            this.dotnetDumpService = dotnetDumpService ?? throw new ArgumentNullException(nameof(dotnetDumpService));
            this.eventConsumer = eventConsumer ?? throw new ArgumentNullException(nameof(eventConsumer));
 
            apiValidationSession = apiValidate;
 
            disposed = false;
            apiValidationStatus = ApiValidationStatus.NeverValidated;
            released = false;
 
            startupTcs = new TaskCompletionSource();
            rebootTcs = new TaskCompletionSource();
            primeTcs = new TaskCompletionSource();
 
            rebootGate = Task.CompletedTask;
            customEventProcessingTask = Task.CompletedTask;
 
            // Run this asynchronously because we want to try to avoid any effects sending topics to the server while the initial bridge request is processing
            // It MAY be the source of a DD crash. See this gist https://gist.github.com/Cyberboss/7776bbeff3a957d76affe0eae95c9f14
            // Worth further investigation as to if that sequence of events is a reliable crash vector and opening a BYOND bug if it is
            initialBridgeRequestTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
            sessionDurationCts = new CancellationTokenSource();
 
            TopicSendSemaphore = new FifoSemaphore();
            synchronizationLock = new object();
 
            if (apiValidationSession || DMApiAvailable)
            {
                bridgeRegistration = bridgeRegistrar.RegisterHandler(this);
                this.chatTrackingContext.SetChannelSink(this);
            }
            else
                logger.LogTrace(
                    "Not registering session with {reasonWhyDmApiIsBad} DMAPI version for interop!",
                    reattachInformation.Dmb.CompileJob.DMApiVersion == null
                        ? "no"
                        : $"incompatible ({reattachInformation.Dmb.CompileJob.DMApiVersion})");
 
            async Task<int?> WrapLifetime()
            {
                var exitCode = await process.Lifetime;
                await postLifetimeCallback();
                if (postValidationShutdownTask != null)
                    await postValidationShutdownTask;
 
                return exitCode;
            }
 
            Lifetime = WrapLifetime();
 
            LaunchResult = GetLaunchResult(
                assemblyInformationProvider,
                asyncDelayer,
                startupTimeout,
                reattached,
                apiValidate);
 
            logger.LogDebug(
                "Created session controller. CommsKey: {accessIdentifier}, Port: {port}",
                reattachInformation.AccessIdentifier,
                reattachInformation.Port);
        }
 
        public async ValueTask DisposeAsync()
        {
            lock (synchronizationLock)
            {
                if (disposed)
                    return;
                disposed = true;
            }
 
            Logger.LogTrace("Disposing...");
 
            sessionDurationCts.Cancel();
            var cancellationToken = CancellationToken.None; // DCT: None available
            var semaphoreLockTask = TopicSendSemaphore.Lock(cancellationToken);
 
            if (!released)
            {
                await engineLock.StopServerProcess(
                    Logger,
                    process,
                    ReattachInformation.AccessIdentifier,
                    ReattachInformation.Port,
                    cancellationToken);
            }
 
            await process.DisposeAsync();
            engineLock.Dispose();
            bridgeRegistration?.Dispose();
            var regularDmbDisposeTask = ReattachInformation.Dmb.DisposeAsync();
            var initialDmb = ReattachInformation.InitialDmb;
            if (initialDmb != null)
                await initialDmb.DisposeAsync();
 
            await regularDmbDisposeTask;
 
            chatTrackingContext.Dispose();
            sessionDurationCts.Dispose();
 
            if (!released)
                await Lifetime; // finish the async callback
 
            (await semaphoreLockTask).Dispose();
            TopicSendSemaphore.Dispose();
 
            await customEventProcessingTask;
        }
 
        public async ValueTask<BridgeResponse?> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken)
        {
            ArgumentNullException.ThrowIfNull(parameters);
 
            using (LogContext.PushProperty(SerilogContextHelper.InstanceIdContextProperty, metadata.Id))
            {
                Logger.LogTrace("Handling bridge request...");
 
                try
                {
                    return await ProcessBridgeCommand(parameters, cancellationToken);
                }
                finally
                {
                    initialBridgeRequestTcs.TrySetResult();
                }
            }
        }
 
        public ValueTask Release()
        {
            CheckDisposed();
 
            ReattachInformation.Dmb.KeepAlive();
            ReattachInformation.InitialDmb?.KeepAlive();
            engineLock.DoNotDeleteThisSession();
            released = true;
            return DisposeAsync();
        }
 
        public ValueTask<TopicResponse?> SendCommand(TopicParameters parameters, CancellationToken cancellationToken)
            => SendCommand(parameters, false, cancellationToken);
 
        public async ValueTask<bool> SetRebootState(RebootState newRebootState, CancellationToken cancellationToken)
        {
            if (RebootState == newRebootState)
                return true;
 
            Logger.LogTrace("Changing reboot state to {newRebootState}", newRebootState);
 
            ReattachInformation.RebootState = newRebootState;
            var result = await SendCommand(
                new TopicParameters(newRebootState),
                cancellationToken);
 
            return result != null && result.ErrorMessage == null;
        }
 
        public void ResetRebootState()
        {
            CheckDisposed();
            Logger.LogTrace("Resetting reboot state...");
            ReattachInformation.RebootState = RebootState.Normal;
        }
 
        public void AdjustPriority(bool higher) => process.AdjustPriority(higher);
 
        public void SuspendProcess() => process.SuspendProcess();
 
        public void ResumeProcess() => process.ResumeProcess();
 
        public IAsyncDisposable ReplaceDmbProvider(IDmbProvider dmbProvider)
        {
            var oldDmb = ReattachInformation.Dmb;
            ReattachInformation.Dmb = dmbProvider ?? throw new ArgumentNullException(nameof(dmbProvider));
            return oldDmb;
        }
 
        public async ValueTask InstanceRenamed(string newInstanceName, CancellationToken cancellationToken)
        {
            var runtimeInformation = ReattachInformation.RuntimeInformation;
            if (runtimeInformation != null)
                runtimeInformation.InstanceName = newInstanceName;
 
            await SendCommand(
                TopicParameters.CreateInstanceRenamedTopicParameters(newInstanceName),
                cancellationToken);
        }
 
        public async ValueTask UpdateChannels(IEnumerable<ChannelRepresentation> newChannels, CancellationToken cancellationToken)
            => await SendCommand(
                new TopicParameters(
                    new ChatUpdate(newChannels)),
                cancellationToken);
 
        public ValueTask CreateDump(string outputFile, bool minidump, CancellationToken cancellationToken)
        {
            if (engineLock.UseDotnetDump)
                return dotnetDumpService.Dump(process, outputFile, minidump, cancellationToken);
 
            return process.CreateDump(outputFile, minidump, cancellationToken);
        }
 
        public double MeasureProcessorTimeDelta()
            => process.MeasureProcessorTimeDelta();
 
        async Task<LaunchResult> GetLaunchResult(
            IAssemblyInformationProvider assemblyInformationProvider,
            IAsyncDelayer asyncDelayer,
            uint? startupTimeout,
            bool reattached,
            bool apiValidate)
        {
            var startTime = DateTimeOffset.UtcNow;
            var useBridgeRequestForLaunchResult = !reattached && (apiValidate || DMApiAvailable);
            var startupTask = useBridgeRequestForLaunchResult
                ? initialBridgeRequestTcs.Task
                : process.Startup;
            var toAwait = Task.WhenAny(startupTask, process.Lifetime);
 
            if (startupTimeout.HasValue)
                toAwait = Task.WhenAny(
                    toAwait,
                    asyncDelayer.Delay(
                        TimeSpan.FromSeconds(startupTimeout.Value),
                        CancellationToken.None)
                    .AsTask()); // DCT: None available, task will clean up after delay
 
            Logger.LogTrace(
                "Waiting for LaunchResult based on {launchResultCompletionCause}{possibleTimeout}...",
                useBridgeRequestForLaunchResult ? "initial bridge request" : "process startup",
                startupTimeout.HasValue ? $" with a timeout of {startupTimeout.Value}s" : String.Empty);
 
            await toAwait;
 
            var result = new LaunchResult
            {
                ExitCode = process.Lifetime.IsCompleted ? await process.Lifetime : null,
                StartupTime = startupTask.IsCompleted ? (DateTimeOffset.UtcNow - startTime) : null,
            };
 
            Logger.LogTrace("Launch result: {launchResult}", result);
 
            if (!result.ExitCode.HasValue && reattached && !disposed)
            {
                var reattachResponse = await SendCommand(
                    new TopicParameters(
                        assemblyInformationProvider.Version,
                        ReattachInformation.RuntimeInformation!.ServerPort),
                    true,
                    sessionDurationCts.Token);
 
                if (reattachResponse != null)
                {
                    if (reattachResponse?.CustomCommands != null)
                        chatTrackingContext.CustomCommands = reattachResponse.CustomCommands;
                    else if (reattachResponse != null)
                        Logger.Log(
                            CompileJob.DMApiVersion >= new Version(5, 2, 0)
                                ? LogLevel.Warning
                                : LogLevel.Debug,
                            "DMAPI Interop v{interopVersion} isn't returning the TGS custom commands list. Functionality added in v5.2.0.",
                            CompileJob.DMApiVersion!.Semver());
                }
            }
 
            return result;
        }
 
        void CheckDisposed() => ObjectDisposedException.ThrowIf(disposed, this);
 
        async Task PostValidationShutdown(Task<bool> proceedTask)
        {
            Logger.LogTrace("Entered post validation terminate task.");
            if (!await proceedTask)
            {
                Logger.LogTrace("Not running post validation terminate task for repeated bridge request.");
                return;
            }
 
            const int GracePeriodSeconds = 30;
            Logger.LogDebug("Server will terminated in {gracePeriodSeconds}s if it does not exit...", GracePeriodSeconds);
            var delayTask = asyncDelayer.Delay(TimeSpan.FromSeconds(GracePeriodSeconds), CancellationToken.None).AsTask(); // DCT: None available
            await Task.WhenAny(process.Lifetime, delayTask);
 
            if (!process.Lifetime.IsCompleted)
            {
                Logger.LogWarning("DMAPI took too long to shutdown server after validation request!");
                process.Terminate();
                apiValidationStatus = ApiValidationStatus.BadValidationRequest;
            }
            else
                Logger.LogTrace("Server exited properly post validation.");
        }
 
#pragma warning disable CA1502 // TODO: Decomplexify
        async ValueTask<BridgeResponse?> ProcessBridgeCommand(BridgeParameters parameters, CancellationToken cancellationToken)
        {
            var response = new BridgeResponse();
            switch (parameters.CommandType)
            {
                case BridgeCommandType.ChatSend:
                    if (parameters.ChatMessage == null)
                        return BridgeError("Missing chatMessage field!");
 
                    if (parameters.ChatMessage.ChannelIds == null)
                        return BridgeError("Missing channelIds field in chatMessage!");
 
                    if (parameters.ChatMessage.ChannelIds.Any(channelIdString => !UInt64.TryParse(channelIdString, out var _)))
                        return BridgeError("Invalid channelIds in chatMessage!");
 
                    if (parameters.ChatMessage.Text == null)
                        return BridgeError("Missing message field in chatMessage!");
 
                    var anyFailed = false;
                    var parsedChannels = parameters.ChatMessage.ChannelIds.Select(
                        channelString =>
                        {
                            anyFailed |= !UInt64.TryParse(channelString, out var channelId);
                            return channelId;
                        });
 
                    if (anyFailed)
                        return BridgeError("Failed to parse channelIds as U64!");
 
                    chat.QueueMessage(
                        parameters.ChatMessage,
                        parsedChannels);
                    break;
                case BridgeCommandType.Prime:
                    Interlocked.Exchange(ref primeTcs, new TaskCompletionSource()).SetResult();
                    break;
                case BridgeCommandType.Kill:
                    Logger.LogInformation("Bridge requested process termination!");
                    chatTrackingContext.Active = false;
                    terminationWasIntentional = true;
                    process.Terminate();
                    break;
                case BridgeCommandType.DeprecatedPortUpdate:
                    return BridgeError("Port switching is no longer supported!");
                case BridgeCommandType.Startup:
                    apiValidationStatus = ApiValidationStatus.BadValidationRequest;
 
                    if (apiValidationSession)
                    {
                        var proceedTcs = new TaskCompletionSource<bool>();
                        var firstValidationRequest = Interlocked.CompareExchange(ref postValidationShutdownTask, PostValidationShutdown(proceedTcs.Task), null) == null;
                        proceedTcs.SetResult(firstValidationRequest);
 
                        if (!firstValidationRequest)
                            return BridgeError("Startup bridge request was repeated!");
                    }
 
                    if (parameters.Version == null)
                        return BridgeError("Missing dmApiVersion field!");
 
                    DMApiVersion = parameters.Version;
 
                    // TODO: When OD figures out how to unite port and topic_port, set an upper version bound on OD for this check
                    if (DMApiVersion.Major != DMApiConstants.InteropVersion.Major
                        || (EngineVersion.Engine == EngineType.OpenDream && DMApiVersion < new Version(5, 7)))
                    {
                        apiValidationStatus = ApiValidationStatus.Incompatible;
                        return BridgeError("Incompatible dmApiVersion!");
                    }
 
                    switch (parameters.MinimumSecurityLevel)
                    {
                        case DreamDaemonSecurity.Ultrasafe:
                            apiValidationStatus = ApiValidationStatus.RequiresUltrasafe;
                            break;
                        case DreamDaemonSecurity.Safe:
                            apiValidationStatus = ApiValidationStatus.RequiresSafe;
                            break;
                        case DreamDaemonSecurity.Trusted:
                            apiValidationStatus = ApiValidationStatus.RequiresTrusted;
                            break;
                        case null:
                            return BridgeError("Missing minimumSecurityLevel field!");
                        default:
                            return BridgeError("Invalid minimumSecurityLevel!");
                    }
 
                    Logger.LogTrace("ApiValidationStatus set to {apiValidationStatus}", apiValidationStatus);
 
                    // we create new runtime info here because of potential .Dmb changes (i think. i forget...)
                    response.RuntimeInformation = new RuntimeInformation(
                        chatTrackingContext,
                        ReattachInformation.Dmb,
                        ReattachInformation.RuntimeInformation!.ServerVersion,
                        ReattachInformation.RuntimeInformation.InstanceName,
                        ReattachInformation.RuntimeInformation.SecurityLevel,
                        ReattachInformation.RuntimeInformation.Visibility,
                        ReattachInformation.RuntimeInformation.ServerPort,
                        ReattachInformation.RuntimeInformation.ApiValidateOnly);
 
                    if (parameters.TopicPort.HasValue)
                    {
                        var newTopicPort = parameters.TopicPort.Value;
                        Logger.LogInformation("Server is requesting use of port {topicPort} for topic communications", newTopicPort);
                        ReattachInformation.TopicPort = newTopicPort;
                    }
 
                    // Load custom commands
                    chatTrackingContext.CustomCommands = parameters.CustomCommands ?? Array.Empty<CustomCommand>();
                    chatTrackingContext.Active = true;
                    Interlocked.Exchange(ref startupTcs, new TaskCompletionSource()).SetResult();
                    break;
                case BridgeCommandType.Reboot:
                    Interlocked.Increment(ref rebootBridgeRequestsProcessing);
                    try
                    {
                        chatTrackingContext.Active = false;
                        Interlocked.Exchange(ref rebootTcs, new TaskCompletionSource()).SetResult();
                        await RebootGate.WaitAsync(cancellationToken);
                    }
                    finally
                    {
                        Interlocked.Decrement(ref rebootBridgeRequestsProcessing);
                    }
 
                    break;
                case BridgeCommandType.Chunk:
                    return await ProcessChunk<BridgeParameters, BridgeResponse>(ProcessBridgeCommand, BridgeError, parameters.Chunk, cancellationToken);
                case BridgeCommandType.Event:
                    return TriggerCustomEvent(parameters.EventInvocation);
                case null:
                    return BridgeError("Missing commandType!");
                default:
                    return BridgeError($"commandType {parameters.CommandType} not supported!");
            }
 
            return response;
        }
#pragma warning restore CA1502
 
        BridgeResponse BridgeError(string message)
        {
            Logger.LogWarning("Bridge request error: {message}", message);
            return new BridgeResponse
            {
                ErrorMessage = message,
            };
        }
 
        async ValueTask<CombinedTopicResponse?> SendTopicRequest(TopicParameters parameters, CancellationToken cancellationToken)
        {
            parameters.AccessIdentifier = ReattachInformation.AccessIdentifier;
 
            var fullCommandString = GenerateQueryString(parameters, out var json);
            if (LogTopicRequests)
                Logger.LogTrace("Topic request: {json}", json);
            var fullCommandByteCount = Encoding.UTF8.GetByteCount(fullCommandString);
            var topicPriority = parameters.IsPriority;
            if (fullCommandByteCount <= DMApiConstants.MaximumTopicRequestLength)
                return await SendRawTopic(fullCommandString, topicPriority, cancellationToken);
 
            var interopChunkingVersion = new Version(5, 6, 0);
            if (ReattachInformation.Dmb.CompileJob.DMApiVersion < interopChunkingVersion)
            {
                Logger.LogWarning(
                    "Cannot send topic request as it is exceeds the single request limit of {limitBytes}B ({actualBytes}B) and requires chunking and the current compile job's interop version must be at least {chunkingVersionRequired}!",
                    DMApiConstants.MaximumTopicRequestLength,
                    fullCommandByteCount,
                    interopChunkingVersion);
                return null;
            }
 
            var payloadId = NextPayloadId;
 
            // AccessIdentifer is just noise in a chunked request
            parameters.AccessIdentifier = null!;
            GenerateQueryString(parameters, out json);
 
            // yes, this straight up ignores unicode, precalculating it is useless when we don't
            // even know if the UTF8 bytes of the url encoded chunk will fit the window until we do said encoding
            var fullPayloadSize = (uint)json.Length;
 
            List<string>? chunkQueryStrings = null;
            for (var chunkCount = 2; chunkQueryStrings == null; ++chunkCount)
            {
                var standardChunkSize = fullPayloadSize / chunkCount;
                var bigChunkSize = standardChunkSize + (fullPayloadSize % chunkCount);
                if (bigChunkSize > DMApiConstants.MaximumTopicRequestLength)
                    continue;
 
                chunkQueryStrings = new List<string>();
                for (var i = 0U; i < chunkCount; ++i)
                {
                    var startIndex = i * standardChunkSize;
                    var subStringLength = Math.Min(
                        fullPayloadSize - startIndex,
                        i == chunkCount - 1
                            ? bigChunkSize
                            : standardChunkSize);
                    var chunkPayload = json.Substring((int)startIndex, (int)subStringLength);
 
                    var chunk = new ChunkData
                    {
                        Payload = chunkPayload,
                        PayloadId = payloadId,
                        SequenceId = i,
                        TotalChunks = (uint)chunkCount,
                    };
 
                    var chunkParameters = new TopicParameters(chunk)
                    {
                        AccessIdentifier = ReattachInformation.AccessIdentifier,
                    };
 
                    var chunkCommandString = GenerateQueryString(chunkParameters, out _);
                    if (Encoding.UTF8.GetByteCount(chunkCommandString) > DMApiConstants.MaximumTopicRequestLength)
                    {
                        // too long when encoded, need more chunks
                        chunkQueryStrings = null;
                        break;
                    }
 
                    chunkQueryStrings.Add(chunkCommandString);
                }
            }
 
            Logger.LogTrace("Chunking topic request ({totalChunks} total)...", chunkQueryStrings.Count);
 
            CombinedTopicResponse? combinedResponse = null;
            bool LogRequestIssue(bool possiblyFromCompletedRequest)
            {
                if (combinedResponse?.InteropResponse == null || combinedResponse.InteropResponse.ErrorMessage != null)
                {
                    Logger.LogWarning(
                        "Topic request {chunkingStatus} failed!{potentialRequestError}",
                        possiblyFromCompletedRequest ? "final chunk" : "chunking",
                        combinedResponse?.InteropResponse?.ErrorMessage != null
                            ? $" Request error: {combinedResponse.InteropResponse.ErrorMessage}"
                            : String.Empty);
                    return true;
                }
 
                return false;
            }
 
            foreach (var chunkCommandString in chunkQueryStrings)
            {
                combinedResponse = await SendRawTopic(chunkCommandString, topicPriority, cancellationToken);
                if (LogRequestIssue(chunkCommandString == chunkQueryStrings.Last()))
                    return null;
            }
 
            while ((combinedResponse?.InteropResponse?.MissingChunks?.Count ?? 0) > 0)
            {
                Logger.LogWarning("DD is still missing some chunks of topic request P{payloadId}! Sending missing chunks...", payloadId);
                var missingChunks = combinedResponse!.InteropResponse!.MissingChunks!;
                var lastIndex = missingChunks.Last();
                foreach (var missingChunkIndex in missingChunks)
                {
                    var chunkCommandString = chunkQueryStrings[(int)missingChunkIndex];
                    combinedResponse = await SendRawTopic(chunkCommandString, topicPriority, cancellationToken);
                    if (LogRequestIssue(missingChunkIndex == lastIndex))
                        return null;
                }
            }
 
            return combinedResponse;
        }
 
        string GenerateQueryString(TopicParameters parameters, out string json)
        {
            json = JsonConvert.SerializeObject(parameters, DMApiConstants.SerializerSettings);
            var commandString = String.Format(
                CultureInfo.InvariantCulture,
                "?{0}={1}",
                byondTopicSender.SanitizeString(DMApiConstants.TopicData),
                byondTopicSender.SanitizeString(json));
            return commandString;
        }
 
        async ValueTask<CombinedTopicResponse?> SendRawTopic(string queryString, bool priority, CancellationToken cancellationToken)
        {
            if (disposed)
            {
                Logger.LogWarning(
                    "Attempted to send a topic on a disposed SessionController");
                return null;
            }
 
            var targetPort = ReattachInformation.TopicPort ?? ReattachInformation.Port;
            Byond.TopicSender.TopicResponse? byondResponse;
            using (await TopicSendSemaphore.Lock(cancellationToken))
                byondResponse = await byondTopicSender.SendWithOptionalPriority(
                    asyncDelayer,
                    LogTopicRequests
                        ? Logger
                        : NullLogger.Instance,
                    queryString,
                    targetPort,
                    priority,
                    cancellationToken);
 
            if (byondResponse == null)
            {
                if (priority)
                    Logger.LogError(
                        "Unable to send priority topic \"{queryString}\"!",
                        queryString);
 
                return null;
            }
 
            var topicReturn = byondResponse.StringData;
 
            TopicResponse? interopResponse = null;
            if (topicReturn != null)
                try
                {
                    interopResponse = JsonConvert.DeserializeObject<TopicResponse>(topicReturn, DMApiConstants.SerializerSettings);
                }
                catch (Exception ex)
                {
                    Logger.LogWarning(ex, "Invalid interop response: {topicReturnString}", topicReturn);
                }
 
            return new CombinedTopicResponse(byondResponse, interopResponse);
        }
 
        async ValueTask<TopicResponse?> SendCommand(TopicParameters parameters, bool bypassLaunchResult, CancellationToken cancellationToken)
        {
            ArgumentNullException.ThrowIfNull(parameters);
 
            if (Lifetime.IsCompleted || disposed)
            {
                Logger.LogWarning(
                    "Attempted to send a command to an inactive SessionController: {commandType}",
                    parameters.CommandType);
                return null;
            }
 
            if (!DMApiAvailable)
            {
                Logger.LogTrace("Not sending topic request {commandType} to server without/with incompatible DMAPI!", parameters.CommandType);
                return null;
            }
 
            var reboot = OnReboot;
            if (!bypassLaunchResult)
            {
                var launchResult = await LaunchResult.WaitAsync(cancellationToken);
                if (launchResult.ExitCode.HasValue)
                {
                    Logger.LogDebug("Not sending topic request {commandType} to server that failed to launch!", parameters.CommandType);
                    return null;
                }
            }
 
            // meh, this is kind of a hack, but it works
            if (!chatTrackingContext.Active)
            {
                Logger.LogDebug("Not sending topic request {commandType} to server that is rebooting/starting.", parameters.CommandType);
                return null;
            }
 
            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            var combinedCancellationToken = cts.Token;
            async ValueTask CancelIfLifetimeElapses()
            {
                try
                {
                    var completed = await Task.WhenAny(Lifetime, reboot).WaitAsync(combinedCancellationToken);
 
                    Logger.LogDebug(
                        "Server {action}, cancelling pending command: {commandType}",
                        completed != reboot
                            ? "process ended"
                            : "rebooting",
                        parameters.CommandType);
                    cts.Cancel();
                }
                catch (OperationCanceledException)
                {
                    // expected, not even worth tracing
                }
                catch (Exception ex)
                {
                    Logger.LogError(ex, "Error in CancelIfLifetimeElapses!");
                }
            }
 
            TopicResponse? fullResponse = null;
            var lifetimeWatchingTask = CancelIfLifetimeElapses();
            try
            {
                var combinedResponse = await SendTopicRequest(parameters, combinedCancellationToken);
 
                void LogCombinedResponse()
                {
                    if (LogTopicRequests && combinedResponse != null)
                        Logger.LogTrace("Topic response: {topicString}", combinedResponse.ByondTopicResponse.StringData ?? "(NO STRING DATA)");
                }
 
                LogCombinedResponse();
 
                if (combinedResponse?.InteropResponse?.Chunk != null)
                {
                    Logger.LogTrace("Topic response is chunked...");
 
                    ChunkData? nextChunk = combinedResponse.InteropResponse.Chunk;
                    do
                    {
                        var nextRequest = await ProcessChunk<TopicResponse, ChunkedTopicParameters>(
                            (completedResponse, _) =>
                            {
                                fullResponse = completedResponse;
                                return ValueTask.FromResult<ChunkedTopicParameters?>(null);
                            },
                            error =>
                            {
                                Logger.LogWarning("Topic response chunking error: {message}", error);
                                return null;
                            },
                            combinedResponse?.InteropResponse?.Chunk,
                            combinedCancellationToken);
 
                        if (nextRequest != null)
                        {
                            nextRequest.PayloadId = nextChunk.PayloadId;
                            combinedResponse = await SendTopicRequest(nextRequest, combinedCancellationToken);
                            LogCombinedResponse();
                            nextChunk = combinedResponse?.InteropResponse?.Chunk;
                        }
                        else
                            nextChunk = null;
                    }
                    while (nextChunk != null);
                }
                else
                    fullResponse = combinedResponse?.InteropResponse;
            }
            catch (OperationCanceledException ex)
            {
                Logger.LogDebug(
                    ex,
                    "Topic request {cancellationType}!",
                    combinedCancellationToken.IsCancellationRequested
                        ? cancellationToken.IsCancellationRequested
                            ? "cancelled"
                            : "aborted"
                        : "timed out");
 
                // throw only if the original token was the trigger
                cancellationToken.ThrowIfCancellationRequested();
            }
            finally
            {
                cts.Cancel();
                await lifetimeWatchingTask;
            }
 
            if (fullResponse?.ErrorMessage != null)
                Logger.LogWarning(
                    "Errored topic response for command {commandType}: {errorMessage}",
                    parameters.CommandType,
                    fullResponse.ErrorMessage);
 
            return fullResponse;
        }
 
        BridgeResponse TriggerCustomEvent(CustomEventInvocation? invocation)
        {
            if (invocation == null)
                return BridgeError("Missing eventInvocation!");
 
            var eventName = invocation.EventName;
            if (eventName == null)
                return BridgeError("Missing eventName!");
 
            var notifyCompletion = invocation.NotifyCompletion;
            if (!notifyCompletion.HasValue)
                return BridgeError("Missing notifyCompletion!");
 
            var eventParams = new List<string>
            {
                ReattachInformation.Dmb.Directory,
            };
 
            eventParams.AddRange(invocation
                .Parameters?
                .Where(param => param != null)
                .Cast<string>()
                ?? Enumerable.Empty<string>());
 
            var eventId = Guid.NewGuid();
            Logger.LogInformation("Triggering custom event \"{eventName}\": {eventId}", eventName, eventId);
 
            var cancellationToken = sessionDurationCts.Token;
            ValueTask? eventTask = eventConsumer.HandleCustomEvent(eventName, eventParams, cancellationToken);
 
            async Task ProcessEvent()
            {
                try
                {
                    Exception? exception;
                    try
                    {
                        await eventTask.Value;
                        exception = null;
                    }
                    catch (Exception ex)
                    {
                        exception = ex;
                    }
 
                    if (notifyCompletion.Value)
                        await SendCommand(
                            new TopicParameters(eventId),
                            cancellationToken);
                    else if (exception == null)
                        Logger.LogTrace("Finished custom event {eventId}, not sending notification.", eventId);
 
                    if (exception != null)
                        throw exception;
                }
                catch (OperationCanceledException ex)
                {
                    Logger.LogDebug(ex, "Custom event invocation {eventId} aborted!", eventId);
                }
                catch (Exception ex)
                {
                    Logger.LogWarning(ex, "Custom event invocation {eventId} errored!", eventId);
                }
            }
 
            if (!eventTask.HasValue)
                return BridgeError("Event refused to execute due to matching a TGS event!");
 
            lock (sessionDurationCts)
            {
                var previousEventProcessingTask = customEventProcessingTask;
                var eventProcessingTask = ProcessEvent();
                customEventProcessingTask = Task.WhenAll(customEventProcessingTask, eventProcessingTask);
            }
 
            return new BridgeResponse
            {
                EventId = notifyCompletion.Value
                    ? eventId.ToString()
                    : null,
            };
        }
    }
}