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.Rights;
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.Jobs;
using Tgstation.Server.Host.Models;
using Tgstation.Server.Host.System;
using Tgstation.Server.Host.Utils;
 
namespace Tgstation.Server.Host.Components.Session
{
#pragma warning disable CA1506
    sealed class SessionController : Chunker, ISessionController, IBridgeHandler, IChannelSink
#pragma warning restore CA1506
    {
        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 => terminationWasIntentionalForced || (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 long? StartupBridgeRequestsReceived { get; private set; }
 
        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 IJobManager jobManager;
 
        readonly IDreamMaker dreamMaker;
 
        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 terminationWasIntentionalForced;
 
        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,
            IJobManager jobManager,
            IDreamMaker dreamMaker,
            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));
            this.jobManager = jobManager ?? throw new ArgumentNullException(nameof(jobManager));
            this.dreamMaker = dreamMaker ?? throw new ArgumentNullException(nameof(dreamMaker));
 
            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 (DMApiAvailable)
            {
                StartupBridgeRequestsReceived = 0;
            }
 
            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))
            {
                if (!DMApiAvailable && !apiValidationSession)
                {
                    Logger.LogWarning("Ignoring bridge request from session without confirmed DMAPI!");
                    return null;
                }
 
                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
#pragma warning disable CA1506
        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;
                    terminationWasIntentionalForced = true;
                    process.Terminate();
                    break;
                case BridgeCommandType.DeprecatedPortUpdate:
                    return BridgeError("Port switching is no longer supported!");
                case BridgeCommandType.Startup:
                    apiValidationStatus = ApiValidationStatus.BadValidationRequest;
                    ++StartupBridgeRequestsReceived;
 
                    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;
                        var rebootGate = RebootGate; // Important to read this before setting the TCS or it could change
                        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 BridgeCommandType.Deploy:
                    var job = Job.Create(JobCode.AutomaticDeployment, null, metadata, DreamMakerRights.CancelCompile);
                    await jobManager.RegisterOperation(
                        job,
                        (core, databaseContextFactory, job, progressReporter, jobCancellationToken) =>
                            dreamMaker.DeploymentProcess(
                                job,
                                databaseContextFactory,
                                progressReporter,
                                jobCancellationToken),
                        cancellationToken);
 
                    break;
                case null:
                    return BridgeError("Missing commandType!");
                default:
                    return BridgeError($"commandType {parameters.CommandType} not supported!");
            }
 
            return response;
        }
#pragma warning restore CA1506
#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,
            };
        }
    }
}