Provider.cs

Go to the documentation of this file.

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
 
using Microsoft.Extensions.Logging;
 
using Tgstation.Server.Api.Rights;
using Tgstation.Server.Host.Components.Interop;
using Tgstation.Server.Host.Jobs;
using Tgstation.Server.Host.Models;
using Tgstation.Server.Host.Utils;
 
namespace Tgstation.Server.Host.Components.Chat.Providers
{
    abstract class Provider : IProvider
    {
        public Task InitialConnectionJob => initialConnectionTcs.Task;
 
        protected ChatBot ChatBot { get; }
 
        protected IAsyncDelayer AsyncDelayer { get; }
 
        protected ILogger<Provider> Logger { get; }
 
        readonly IJobManager jobManager;
 
        readonly Queue<Message?> messageQueue;
 
        readonly TaskCompletionSource initialConnectionTcs;
 
        readonly object reconnectTaskLock;
 
        TaskCompletionSource nextMessage;
 
        Task? reconnectTask;
 
        CancellationTokenSource? reconnectCts;
 
        protected static string GetEngineCompilerPrefix(Api.Models.EngineType engineType)
            => engineType switch
            {
                Api.Models.EngineType.Byond => "DM",
                Api.Models.EngineType.OpenDream => "OD",
                _ => throw new InvalidOperationException($"Unsupported engine type: {engineType}"),
            };
 
        protected Provider(IJobManager jobManager, IAsyncDelayer asyncDelayer, ILogger<Provider> logger, ChatBot chatBot)
        {
            this.jobManager = jobManager ?? throw new ArgumentNullException(nameof(jobManager));
            AsyncDelayer = asyncDelayer ?? throw new ArgumentNullException(nameof(asyncDelayer));
            Logger = logger ?? throw new ArgumentNullException(nameof(logger));
            ChatBot = chatBot ?? throw new ArgumentNullException(nameof(chatBot));
 
            if (chatBot.Instance == null)
                throw new ArgumentException("chatBot must have Instance!", nameof(chatBot));
 
            messageQueue = new Queue<Message?>();
            nextMessage = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
            initialConnectionTcs = new TaskCompletionSource();
            reconnectTaskLock = new object();
 
            logger.LogTrace("Created.");
        }
 
        public abstract bool Connected { get; }
 
        public abstract string BotMention { get; }
 
        public bool Disposed { get; private set; }
 
        public virtual async ValueTask DisposeAsync()
        {
            Disposed = true;
            await StopReconnectionTimer();
 
            // queue a final message to shutdown the NextMessage Task
            EnqueueMessage(null);
            Logger.LogTrace("Disposed");
        }
 
        public async ValueTask Disconnect(CancellationToken cancellationToken)
        {
            await StopReconnectionTimer();
 
            if (Connected)
            {
                Logger.LogTrace("Disconnecting...");
                await DisconnectImpl(cancellationToken);
                Logger.LogTrace("Disconnected");
            }
        }
 
        public void InitialMappingComplete() => initialConnectionTcs.TrySetResult();
 
        public async ValueTask<Dictionary<ChatChannel, IEnumerable<ChannelRepresentation>>> MapChannels(IEnumerable<ChatChannel> channels, CancellationToken cancellationToken)
        {
            ArgumentNullException.ThrowIfNull(channels);
 
            try
            {
                return await MapChannelsImpl(channels, cancellationToken);
            }
            catch
            {
                initialConnectionTcs.TrySetResult();
                throw;
            }
        }
 
        public async Task<Message?> NextMessage(CancellationToken cancellationToken)
        {
            while (true)
            {
                await nextMessage.Task.WaitAsync(cancellationToken);
                lock (messageQueue)
                    if (messageQueue.Count > 0)
                    {
                        var result = messageQueue.Dequeue();
                        if (messageQueue.Count == 0)
                            nextMessage = new TaskCompletionSource();
                        return result;
                    }
            }
        }
 
        public Task SetReconnectInterval(uint reconnectInterval, bool connectNow)
        {
            if (reconnectInterval == 0)
                throw new ArgumentOutOfRangeException(nameof(reconnectInterval), reconnectInterval, "Reconnect interval cannot be zero!");
 
            Task stopOldTimerTask;
            lock (reconnectTaskLock)
            {
                stopOldTimerTask = StopReconnectionTimer();
                reconnectCts = new CancellationTokenSource();
                reconnectTask = ReconnectionLoop(reconnectInterval, connectNow, reconnectCts.Token);
            }
 
            return stopOldTimerTask;
        }
 
        public abstract ValueTask SendMessage(Message? replyTo, MessageContent message, ulong channelId, CancellationToken cancellationToken);
 
        public abstract ValueTask<Func<string?, string, ValueTask<Func<bool, ValueTask>>>> SendUpdateMessage(
            RevisionInformation revisionInformation,
            RevisionInformation? previousRevisionInformation,
            Api.Models.EngineVersion engineVersion,
            DateTimeOffset? estimatedCompletionTime,
            string? gitHubOwner,
            string? gitHubRepo,
            ulong channelId,
            bool localCommitPushed,
            CancellationToken cancellationToken);
 
        protected abstract ValueTask Connect(CancellationToken cancellationToken);
 
        protected abstract ValueTask DisconnectImpl(CancellationToken cancellationToken);
 
        protected abstract ValueTask<Dictionary<ChatChannel, IEnumerable<ChannelRepresentation>>> MapChannelsImpl(
            IEnumerable<ChatChannel> channels,
            CancellationToken cancellationToken);
 
        protected void EnqueueMessage(Message? message)
        {
            if (message == null)
                Logger.LogTrace("Requesting channel remap...");
 
            lock (messageQueue)
            {
                messageQueue.Enqueue(message);
                nextMessage.TrySetResult();
            }
        }
 
        Task StopReconnectionTimer()
        {
            Logger.LogTrace("StopReconnectionTimer");
            lock (reconnectTaskLock)
                if (reconnectCts != null)
                {
                    reconnectCts.Cancel();
                    reconnectCts.Dispose();
                    reconnectCts = null;
                    var reconnectTask = this.reconnectTask!;
                    this.reconnectTask = null;
                    return reconnectTask;
                }
                else
                    Logger.LogTrace("Timer wasn't running");
 
            return Task.CompletedTask;
        }
 
        async Task ReconnectionLoop(uint reconnectInterval, bool connectNow, CancellationToken cancellationToken)
        {
            do
            {
                try
                {
                    if (!connectNow)
                        await AsyncDelayer.Delay(TimeSpan.FromMinutes(reconnectInterval), cancellationToken);
                    else
                        connectNow = false;
                    if (!Connected)
                    {
                        var job = Job.Create(Api.Models.JobCode.ReconnectChatBot, null, ChatBot.Instance!, ChatBotRights.WriteEnabled);
                        job.Description += $": {ChatBot.Name}";
 
                        await jobManager.RegisterOperation(
                            job,
                            async (core, databaseContextFactory, paramJob, progressReporter, jobCancellationToken) =>
                            {
                                try
                                {
                                    if (Connected)
                                    {
                                        Logger.LogTrace("Disconnecting...");
                                        await DisconnectImpl(jobCancellationToken);
                                    }
                                    else
                                        Logger.LogTrace("Already disconnected not doing disconnection attempt!");
 
                                    Logger.LogTrace("Connecting...");
                                    await Connect(jobCancellationToken);
                                    Logger.LogTrace("Connected successfully");
                                    EnqueueMessage(null);
                                }
                                catch
                                {
                                    // we set this here because otherwise there could be stuff waiting on to connect us forever
                                    initialConnectionTcs.TrySetResult();
                                    throw;
                                }
                            },
                            cancellationToken);
 
                        await jobManager.WaitForJobCompletion(job, null, cancellationToken, cancellationToken);
                    }
                }
                catch (OperationCanceledException e)
                {
                    Logger.LogTrace(e, "ReconnectionLoop cancelled");
                }
                catch (Exception e)
                {
                    Logger.LogError(e, "Error reconnecting!");
                }
            }
            while (!cancellationToken.IsCancellationRequested);
 
            Logger.LogTrace("ReconnectionLoop exiting...");
        }
    }
}