Very naughty merge of 'V6' into OpenDream

Doing some very bad cleanups for a merge in here like renaming a few fields (like the old column names) + `ByondCommand` + Removing deprecated API fields.

src/Tgstation.Server.Client/ApiClient.cs

@@ -11,6 +11,10 @@ using System.Threading;
 using System.Threading.Tasks;
 using System.Web;
 
+using Microsoft.AspNetCore.Http.Connections;
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Microsoft.Net.Http.Headers;
 
 using Newtonsoft.Json;
@@ -20,6 +24,7 @@ using Newtonsoft.Json.Serialization;
 using Tgstation.Server.Api;
 using Tgstation.Server.Api.Models;
 using Tgstation.Server.Api.Models.Response;
+using Tgstation.Server.Client.Extensions;
 using Tgstation.Server.Common.Extensions;
 using Tgstation.Server.Common.Http;
 
@@ -51,6 +56,18 @@ namespace Tgstation.Server.Client
 			set => httpClient.Timeout = value;
 		}
 
+		/// <summary>
+		/// The <see cref="JsonSerializerSettings"/> to use.
+		/// </summary>
+		static readonly JsonSerializerSettings SerializerSettings = new ()
+		{
+			ContractResolver = new CamelCasePropertyNamesContractResolver(),
+			Converters = new[]
+			{
+				new VersionConverter(),
+			},
+		};
+
 		/// <summary>
 		/// The <see cref="IHttpClient"/> for the <see cref="ApiClient"/>.
 		/// </summary>
@@ -61,6 +78,11 @@ namespace Tgstation.Server.Client
 		/// </summary>
 		readonly List<IRequestLogger> requestLoggers;
 
+		/// <summary>
+		/// List of <see cref="HubConnection"/>s created by the <see cref="ApiClient"/>.
+		/// </summary>
+		readonly List<HubConnection> hubConnections;
+
 		/// <summary>
 		/// Backing field for <see cref="Headers"/>.
 		/// </summary>
@@ -82,14 +104,9 @@ namespace Tgstation.Server.Client
 		ApiHeaders headers;
 
 		/// <summary>
-		/// Get the <see cref="JsonSerializerSettings"/> to use.
+		/// If the <see cref="ApiClient"/> is disposed.
 		/// </summary>
-		/// <returns>A new <see cref="JsonSerializerSettings"/> instance.</returns>
-		static JsonSerializerSettings GetSerializerSettings() => new ()
-		{
-			ContractResolver = new CamelCasePropertyNamesContractResolver(),
-			Converters = new[] { new VersionConverter() },
-		};
+		bool disposed;
 
 		/// <summary>
 		/// Handle a bad HTTP <paramref name="response"/>.
@@ -102,7 +119,7 @@ namespace Tgstation.Server.Client
 			try
 			{
 				// check if json serializes to an error message
-				errorMessage = JsonConvert.DeserializeObject<ErrorMessageResponse>(json, GetSerializerSettings());
+				errorMessage = JsonConvert.DeserializeObject<ErrorMessageResponse>(json, SerializerSettings);
 			}
 			catch (JsonException)
 			{
@@ -149,7 +166,12 @@ namespace Tgstation.Server.Client
 		/// <param name="apiHeaders">The value of <see cref="Headers"/>.</param>
 		/// <param name="tokenRefreshHeaders">The value of <see cref="tokenRefreshHeaders"/>.</param>
 		/// <param name="authless">The value of <see cref="authless"/>.</param>
-		public ApiClient(IHttpClient httpClient, Uri url, ApiHeaders apiHeaders, ApiHeaders? tokenRefreshHeaders, bool authless)
+		public ApiClient(
+			IHttpClient httpClient,
+			Uri url,
+			ApiHeaders apiHeaders,
+			ApiHeaders? tokenRefreshHeaders,
+			bool authless)
 		{
 			this.httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
 			Url = url ?? throw new ArgumentNullException(nameof(url));
@@ -158,12 +180,27 @@ namespace Tgstation.Server.Client
 			this.authless = authless;
 
 			requestLoggers = new List<IRequestLogger>();
+			hubConnections = new List<HubConnection>();
 			semaphoreSlim = new SemaphoreSlim(1);
 		}
 
 		/// <inheritdoc />
-		public void Dispose()
+		public async ValueTask DisposeAsync()
 		{
+			List<HubConnection> localHubConnections;
+			lock (hubConnections)
+			{
+				if (disposed)
+					return;
+
+				disposed = true;
+
+				localHubConnections = hubConnections.ToList();
+				hubConnections.Clear();
+			}
+
+			await ValueTaskExtensions.WhenAll(hubConnections.Select(connection => connection.DisposeAsync()));
+
 			httpClient.Dispose();
 			semaphoreSlim.Dispose();
 		}
@@ -294,6 +331,130 @@ namespace Tgstation.Server.Client
 			}
 		}
 
+		/// <summary>
+		/// Attempt to refresh the stored Bearer token in <see cref="Headers"/>.
+		/// </summary>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in <see langword="true"/> if the refresh was successful, <see langword="false"/> if a refresh is unable to be performed.</returns>
+		public async ValueTask<bool> RefreshToken(CancellationToken cancellationToken)
+		{
+			if (tokenRefreshHeaders == null)
+				return false;
+
+			var startingToken = headers.Token;
+			await semaphoreSlim.WaitAsync(cancellationToken).ConfigureAwait(false);
+			try
+			{
+				if (startingToken != headers.Token)
+					return true;
+
+				var token = await RunRequest<object, TokenResponse>(Routes.Root, new object(), HttpMethod.Post, null, true, cancellationToken).ConfigureAwait(false);
+				headers = new ApiHeaders(headers.UserAgent!, token);
+			}
+			finally
+			{
+				semaphoreSlim.Release();
+			}
+
+			return true;
+		}
+
+		/// <inheritdoc />
+		public async ValueTask<IAsyncDisposable> CreateHubConnection<THubImplementation>(
+			THubImplementation hubImplementation,
+			IRetryPolicy? retryPolicy,
+			Action<ILoggingBuilder>? loggingConfigureAction,
+			CancellationToken cancellationToken)
+			where THubImplementation : class
+		{
+			if (hubImplementation == null)
+				throw new ArgumentNullException(nameof(hubImplementation));
+
+			retryPolicy ??= new InfiniteThirtySecondMaxRetryPolicy();
+
+			var wrappedPolicy = new ApiClientTokenRefreshRetryPolicy(this, retryPolicy);
+
+			HubConnection? hubConnection = null;
+			var hubConnectionBuilder = new HubConnectionBuilder()
+				.AddNewtonsoftJsonProtocol(options =>
+				{
+					options.PayloadSerializerSettings = SerializerSettings;
+				})
+				.WithAutomaticReconnect(wrappedPolicy)
+				.WithUrl(
+					new Uri(Url, Routes.JobsHub),
+					HttpTransportType.ServerSentEvents,
+					options =>
+					{
+						options.AccessTokenProvider = async () =>
+						{
+							// DCT: None available.
+							if (Headers.Token == null
+								|| (Headers.Token.ParseJwt().ValidTo <= DateTime.UtcNow
+								&& !await RefreshToken(CancellationToken.None)))
+							{
+								_ = hubConnection!.StopAsync(); // DCT: None available.
+								return null;
+							}
+
+							return Headers.Token.Bearer;
+						};
+
+						options.CloseTimeout = Timeout;
+
+						Headers.SetHubConnectionHeaders(options.Headers);
+					});
+
+			if (loggingConfigureAction != null)
+				hubConnectionBuilder.ConfigureLogging(loggingConfigureAction);
+
+			hubConnection = hubConnectionBuilder.Build();
+			try
+			{
+				hubConnection.Closed += async (error) =>
+				{
+					if (error is HttpRequestException httpRequestException)
+					{
+						// .StatusCode isn't in netstandard but fuck the police
+						var property = error.GetType().GetProperty("StatusCode");
+						if (property != null)
+						{
+							var statusCode = (HttpStatusCode?)property.GetValue(error);
+							if (statusCode == HttpStatusCode.Unauthorized
+								&& !await RefreshToken(CancellationToken.None))
+								_ = hubConnection!.StopAsync();
+						}
+					}
+				};
+
+				hubConnection.ProxyOn(hubImplementation);
+
+				Task startTask;
+				lock (hubConnections)
+				{
+					if (disposed)
+						throw new ObjectDisposedException(nameof(ApiClient));
+
+					hubConnections.Add(hubConnection);
+					startTask = hubConnection.StartAsync(cancellationToken);
+				}
+
+				await startTask;
+
+				return hubConnection;
+			}
+			catch
+			{
+				bool needsDispose;
+				lock (hubConnections)
+					needsDispose = hubConnections.Remove(hubConnection);
+
+				if (needsDispose)
+					await hubConnection.DisposeAsync();
+				throw;
+			}
+		}
+
 		/// <summary>
 		/// Main request method.
 		/// </summary>
@@ -305,6 +466,7 @@ namespace Tgstation.Server.Client
 		/// <param name="tokenRefresh">If this is a token refresh operation.</param>
 		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
 		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in the response on success.</returns>
+#pragma warning disable CA1506 // TODO: Decomplexify
 		protected virtual async ValueTask<TResult> RunRequest<TResult>(
 			string route,
 			HttpContent? content,
@@ -320,9 +482,12 @@ namespace Tgstation.Server.Client
 			if (content == null && (method == HttpMethod.Post || method == HttpMethod.Put))
 				throw new InvalidOperationException("content cannot be null for POST or PUT!");
 
+			if (disposed)
+				throw new ObjectDisposedException(nameof(ApiClient));
+
 			HttpResponseMessage response;
 			var fullUri = new Uri(Url, route);
-			var serializerSettings = GetSerializerSettings();
+			var serializerSettings = SerializerSettings;
 			var fileDownload = typeof(TResult) == typeof(Stream);
 			using (var request = new HttpRequestMessage(method, fullUri))
 			{
@@ -336,6 +501,21 @@ namespace Tgstation.Server.Client
 
 					if (authless)
 						request.Headers.Remove(HeaderNames.Authorization);
+					else
+					{
+						var bearer = headersToUse.Token?.Bearer;
+						if (bearer != null)
+						{
+							var parsed = headersToUse.Token!.ParseJwt();
+							var nbf = parsed.ValidFrom;
+							var now = DateTime.UtcNow;
+							if (nbf >= now)
+							{
+								var delay = (nbf - now).Add(TimeSpan.FromMilliseconds(1));
+								await Task.Delay(delay, cancellationToken);
+							}
+						}
+					}
 
 					if (fileDownload)
 						request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(MediaTypeNames.Application.Octet));
@@ -392,38 +572,7 @@ namespace Tgstation.Server.Client
 				}
 			}
 		}
-
-		/// <summary>
-		/// Attempt to refresh the bearer token in the <see cref="headers"/>.
-		/// </summary>
-		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
-		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in <see langword="true"/> if the refresh was successful, <see langword="false"/> otherwise.</returns>
-		async ValueTask<bool> RefreshToken(CancellationToken cancellationToken)
-		{
-			if (tokenRefreshHeaders == null)
-				return false;
-
-			var startingToken = headers.Token;
-			await semaphoreSlim.WaitAsync(cancellationToken).ConfigureAwait(false);
-			try
-			{
-				if (startingToken != headers.Token)
-					return true;
-
-				var token = await RunRequest<object, TokenResponse>(Routes.Root, new object(), HttpMethod.Post, null, true, cancellationToken).ConfigureAwait(false);
-				headers = new ApiHeaders(headers.UserAgent!, token.Bearer!);
-			}
-			catch (ClientException)
-			{
-				return false;
-			}
-			finally
-			{
-				semaphoreSlim.Release();
-			}
-
-			return true;
-		}
+#pragma warning restore CA1506
 
 		/// <summary>
 		/// Main request method.
@@ -449,7 +598,7 @@ namespace Tgstation.Server.Client
 			HttpContent? content = null;
 			if (body != null)
 				content = new StringContent(
-					JsonConvert.SerializeObject(body, typeof(TBody), Formatting.None, GetSerializerSettings()),
+					JsonConvert.SerializeObject(body, typeof(TBody), Formatting.None, SerializerSettings),
 					Encoding.UTF8,
 					ApiHeaders.ApplicationJsonMime);
 

src/Tgstation.Server.Client/ApiClientTokenRefreshRetryPolicy.cs

@@ -0,0 +1,61 @@
+using System;
+using System.Threading;
+
+using Microsoft.AspNetCore.SignalR.Client;
+
+namespace Tgstation.Server.Client
+{
+	/// <summary>
+	/// A <see cref="IRetryPolicy"/> that attempts to refresh a given <see cref="apiClient"/>'s token on the first disconnect.
+	/// </summary>
+	sealed class ApiClientTokenRefreshRetryPolicy : IRetryPolicy
+	{
+		/// <summary>
+		/// The backing <see cref="ApiClient"/>.
+		/// </summary>
+		readonly ApiClient apiClient;
+
+		/// <summary>
+		/// The wrapped <see cref="IRetryPolicy"/>.
+		/// </summary>
+		readonly IRetryPolicy wrappedPolicy;
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="ApiClientTokenRefreshRetryPolicy"/> class.
+		/// </summary>
+		/// <param name="apiClient">The value of <see cref="apiClient"/>.</param>
+		/// <param name="wrappedPolicy">The value of <see cref="wrappedPolicy"/>.</param>
+		public ApiClientTokenRefreshRetryPolicy(ApiClient apiClient, IRetryPolicy wrappedPolicy)
+		{
+			this.apiClient = apiClient ?? throw new ArgumentNullException(nameof(apiClient));
+			this.wrappedPolicy = wrappedPolicy ?? throw new ArgumentNullException(nameof(wrappedPolicy));
+		}
+
+		/// <inheritdoc />
+		public TimeSpan? NextRetryDelay(RetryContext retryContext)
+		{
+			if (retryContext == null)
+				throw new ArgumentNullException(nameof(retryContext));
+
+			if (retryContext.PreviousRetryCount == 0)
+				AttemptTokenRefresh();
+
+			return wrappedPolicy.NextRetryDelay(retryContext);
+		}
+
+		/// <summary>
+		/// Attempt to refresh the <see cref="apiClient"/>s token asynchronously.
+		/// </summary>
+		async void AttemptTokenRefresh()
+		{
+			try
+			{
+				await apiClient.RefreshToken(CancellationToken.None);
+			}
+			catch
+			{
+				// intentionally ignored
+			}
+		}
+	}
+}

src/Tgstation.Server.Client/Extensions/HubConnectionExtensions.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR.Client;
+
+namespace Tgstation.Server.Client.Extensions
+{
+	/// <summary>
+	/// Extension methods for the <see cref="HubConnection"/> <see langword="class"/>.
+	/// </summary>
+	static class HubConnectionExtensions
+	{
+		/// <summary>
+		/// Apply a given <paramref name="proxy"/> to a given <paramref name="hubConnection"/>.
+		/// </summary>
+		/// <typeparam name="TClientProxy">The strongly typed client proxy.</typeparam>
+		/// <param name="hubConnection">The <see cref="HubConnection"/> to proxy on.</param>
+		/// <param name="proxy">The <typeparamref name="TClientProxy"/> to forward operations to.</param>
+		public static void ProxyOn<TClientProxy>(this HubConnection hubConnection, TClientProxy proxy)
+			where TClientProxy : class
+		{
+			if (hubConnection == null)
+				throw new ArgumentNullException(nameof(hubConnection));
+
+			if (proxy == null)
+				throw new ArgumentNullException(nameof(proxy));
+
+			ProxyOn(hubConnection, typeof(TClientProxy), proxy);
+		}
+
+		/// <summary>
+		/// Apply a given <paramref name="proxyObject"/> to a given <paramref name="hubConnection"/>.
+		/// </summary>
+		/// <param name="hubConnection">The <see cref="HubConnection"/> to proxy on.</param>
+		/// <param name="proxyType">The <see cref="Type"/> of <paramref name="proxyObject"/>.</param>
+		/// <param name="proxyObject">The <see cref="object"/> to forward operations to.</param>
+		static void ProxyOn(this HubConnection hubConnection, Type proxyType, object proxyObject)
+		{
+			var clientMethods = proxyType.GetMethods();
+			var cancellationTokenType = typeof(CancellationToken);
+			foreach (var clientMethod in clientMethods)
+			{
+				var parametersList = clientMethod
+					.GetParameters()
+					.Select(parameterInfo => parameterInfo.ParameterType)
+					.ToList();
+
+				var cancellationTokenIndex = parametersList.IndexOf(cancellationTokenType);
+				if (cancellationTokenIndex != -1)
+				{
+					parametersList.RemoveAt(cancellationTokenIndex);
+#if DEBUG
+					if (parametersList.IndexOf(cancellationTokenType) != -1)
+						throw new InvalidOperationException("Cannot ProxyOn a method with multiple CancellationToken parameters!");
+#endif
+				}
+
+				var parameters = parametersList.ToArray();
+
+				object?[] AddCancellationTokenToParametersArray(object?[] parametersArray)
+				{
+					if (cancellationTokenIndex == -1)
+						return parametersArray;
+
+					var newList = parametersArray.ToList();
+					newList.Insert(cancellationTokenIndex, CancellationToken.None);
+					return newList.ToArray();
+				}
+
+				var returnType = clientMethod.ReturnType;
+				if (returnType != typeof(Task))
+				{
+					if (returnType.BaseType != typeof(Task))
+						throw new InvalidOperationException($"Return type {returnType} of {proxyType.FullName}.{clientMethod.Name} is not supported! Only Task and derivatives are supported.");
+
+					var resultProperty = returnType.GetProperty(nameof(Task<object>.Result));
+					hubConnection.On(
+						clientMethod.Name,
+						parameters,
+						async (parameterArray, _) =>
+						{
+							var task = (Task)clientMethod.Invoke(proxyObject, AddCancellationTokenToParametersArray(parameterArray));
+							await task;
+							return resultProperty.GetValue(task);
+						},
+						hubConnection);
+				}
+				else
+					hubConnection.On(
+						clientMethod.Name,
+						parameters,
+						(parameterArray) =>
+						{
+							return (Task)clientMethod.Invoke(proxyObject, AddCancellationTokenToParametersArray(parameterArray));
+						});
+			}
+
+			foreach (var inheritedInterface in proxyType.GetInterfaces())
+				ProxyOn(hubConnection, inheritedInterface, proxyObject);
+		}
+	}
+}

src/Tgstation.Server.Client/IApiClient.cs

@@ -3,6 +3,9 @@ using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.Logging;
+
 using Tgstation.Server.Api;
 using Tgstation.Server.Api.Models;
 using Tgstation.Server.Api.Models.Response;
@@ -12,7 +15,7 @@ namespace Tgstation.Server.Client
 	/// <summary>
 	/// Web interface for the API.
 	/// </summary>
-	interface IApiClient : IDisposable
+	interface IApiClient : IAsyncDisposable
 	{
 		/// <summary>
 		/// The <see cref="ApiHeaders"/> the <see cref="IApiClient"/> uses.
@@ -35,6 +38,22 @@ namespace Tgstation.Server.Client
 		/// <param name="requestLogger">The <see cref="IRequestLogger"/> to add.</param>
 		void AddRequestLogger(IRequestLogger requestLogger);
 
+		/// <summary>
+		/// Subscribe to all job updates available to the <see cref="IServerClient"/>.
+		/// </summary>
+		/// <typeparam name="THubImplementation">The <see cref="Type"/> of the hub being implemented.</typeparam>
+		/// <param name="hubImplementation">The <typeparamref name="THubImplementation"/> to use for proxying the methods of the hub connection.</param>
+		/// <param name="retryPolicy">The optional <see cref="IRetryPolicy"/> to use for the backing connection. The default retry policy waits for 1, 2, 4, 8, and 16 seconds, then 30s repeatedly.</param>
+		/// <param name="loggingConfigureAction">The optional <see cref="Action{T1}"/> used to configure a <see cref="ILoggingBuilder"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>An <see cref="IAsyncDisposable"/> representing the lifetime of the subscription.</returns>
+		ValueTask<IAsyncDisposable> CreateHubConnection<THubImplementation>(
+			THubImplementation hubImplementation,
+			IRetryPolicy? retryPolicy,
+			Action<ILoggingBuilder>? loggingConfigureAction,
+			CancellationToken cancellationToken)
+			where THubImplementation : class;
+
 		/// <summary>
 		/// Run an HTTP PUT request.
 		/// </summary>

src/Tgstation.Server.Client/IApiClientFactory.cs

@@ -17,6 +17,10 @@ namespace Tgstation.Server.Client
 		/// <param name="tokenRefreshHeaders">The <see cref="ApiHeaders"/> to use to generate a new <see cref="Api.Models.Response.TokenResponse"/>.</param>
 		/// <param name="authless">If there should be no authentication performed.</param>
 		/// <returns>A new <see cref="IApiClient"/>.</returns>
-		IApiClient CreateApiClient(Uri url, ApiHeaders apiHeaders, ApiHeaders? tokenRefreshHeaders, bool authless);
+		IApiClient CreateApiClient(
+			Uri url,
+			ApiHeaders apiHeaders,
+			ApiHeaders? tokenRefreshHeaders,
+			bool authless);
 	}
 }

src/Tgstation.Server.Client/IServerClient.cs

@@ -2,6 +2,10 @@
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.Logging;
+
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Api.Models.Response;
 
 namespace Tgstation.Server.Client
@@ -9,7 +13,7 @@ namespace Tgstation.Server.Client
 	/// <summary>
 	/// Main client for communicating with a server.
 	/// </summary>
-	public interface IServerClient : IDisposable
+	public interface IServerClient : IAsyncDisposable
 	{
 		/// <summary>
 		/// The connected server <see cref="Uri"/>.
@@ -53,6 +57,20 @@ namespace Tgstation.Server.Client
 		/// <returns>A <see cref="Task{TResult}"/> resulting in the <see cref="ServerInformationResponse"/> of the target server.</returns>
 		ValueTask<ServerInformationResponse> ServerInformation(CancellationToken cancellationToken);
 
+		/// <summary>
+		/// Subscribe to all job updates available to the <see cref="IServerClient"/>.
+		/// </summary>
+		/// <param name="jobsReceiver">The <see cref="IJobsHub"/> to use to subscribe to updates.</param>
+		/// <param name="retryPolicy">The optional <see cref="IRetryPolicy"/> to use for the backing connection. The default retry policy waits for 1, 2, 4, 8, and 16 seconds, then 30s repeatedly.</param>
+		/// <param name="loggingConfigureAction">The optional <see cref="Action{T1}"/> used to configure a <see cref="ILoggingBuilder"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>An <see cref="IAsyncDisposable"/> representing the lifetime of the subscription.</returns>
+		ValueTask<IAsyncDisposable> SubscribeToJobUpdates(
+			IJobsHub jobsReceiver,
+			IRetryPolicy? retryPolicy = null,
+			Action<ILoggingBuilder>? loggingConfigureAction = null,
+			CancellationToken cancellationToken = default);
+
 		/// <summary>
 		/// Adds a <paramref name="requestLogger"/> to the request pipeline.
 		/// </summary>

src/Tgstation.Server.Client/InfiniteThirtySecondMaxRetryPolicy.cs

@@ -0,0 +1,21 @@
+using System;
+
+using Microsoft.AspNetCore.SignalR.Client;
+
+namespace Tgstation.Server.Client
+{
+	/// <summary>
+	/// A <see cref="IRetryPolicy"/> that returns seconds in powers of 2, maxing out at 30s.
+	/// </summary>
+	sealed class InfiniteThirtySecondMaxRetryPolicy : IRetryPolicy
+	{
+		/// <inheritdoc />
+		public TimeSpan? NextRetryDelay(RetryContext retryContext)
+		{
+			if (retryContext == null)
+				throw new ArgumentNullException(nameof(retryContext));
+
+			return TimeSpan.FromSeconds(Math.Min(Math.Pow(2, retryContext.PreviousRetryCount), 30));
+		}
+	}
+}

src/Tgstation.Server.Client/ServerClient.cs

@@ -2,7 +2,11 @@
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.Logging;
+
 using Tgstation.Server.Api;
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Api.Models.Response;
 
 namespace Tgstation.Server.Client
@@ -16,12 +20,8 @@ namespace Tgstation.Server.Client
 		/// <inheritdoc />
 		public TokenResponse Token
 		{
-			get => token;
-			set
-			{
-				token = value ?? throw new InvalidOperationException("Cannot set a null Token!");
-				apiClient.Headers = new ApiHeaders(apiClient.Headers.UserAgent!, token.Bearer!);
-			}
+			get => apiClient.Headers.Token ?? throw new InvalidOperationException("apiClient.Headers.Token was null!");
+			set => apiClient.Headers = new ApiHeaders(apiClient.Headers.UserAgent!, value);
 		}
 
 		/// <inheritdoc />
@@ -48,23 +48,13 @@ namespace Tgstation.Server.Client
 		/// </summary>
 		readonly IApiClient apiClient;
 
-		/// <summary>
-		/// Backing field for <see cref="Token"/>.
-		/// </summary>
-		TokenResponse token;
-
 		/// <summary>
 		/// Initializes a new instance of the <see cref="ServerClient"/> class.
 		/// </summary>
 		/// <param name="apiClient">The value of <see cref="apiClient"/>.</param>
-		/// <param name="token">The value of <see cref="Token"/>.</param>
-		public ServerClient(IApiClient apiClient, TokenResponse token)
+		public ServerClient(IApiClient apiClient)
 		{
 			this.apiClient = apiClient ?? throw new ArgumentNullException(nameof(apiClient));
-			this.token = token ?? throw new ArgumentNullException(nameof(token));
-
-			if (Token.Bearer != apiClient.Headers.Token)
-				throw new ArgumentOutOfRangeException(nameof(token), token, "Provided token does not match apiClient headers!");
 
 			Instances = new InstanceManagerClient(apiClient);
 			Users = new UsersClient(apiClient);
@@ -73,12 +63,20 @@ namespace Tgstation.Server.Client
 		}
 
 		/// <inheritdoc />
-		public void Dispose() => apiClient.Dispose();
+		public ValueTask DisposeAsync() => apiClient.DisposeAsync();
 
 		/// <inheritdoc />
 		public ValueTask<ServerInformationResponse> ServerInformation(CancellationToken cancellationToken) => apiClient.Read<ServerInformationResponse>(Routes.Root, cancellationToken);
 
 		/// <inheritdoc />
 		public void AddRequestLogger(IRequestLogger requestLogger) => apiClient.AddRequestLogger(requestLogger);
+
+		/// <inheritdoc />
+		public ValueTask<IAsyncDisposable> SubscribeToJobUpdates(
+			IJobsHub jobsReceiver,
+			IRetryPolicy? retryPolicy,
+			Action<ILoggingBuilder>? loggingConfigureAction,
+			CancellationToken cancellationToken)
+			=> apiClient.CreateHubConnection(jobsReceiver, retryPolicy, loggingConfigureAction, cancellationToken);
 	}
 }

src/Tgstation.Server.Client/ServerClientFactory.cs

@@ -102,7 +102,15 @@ namespace Tgstation.Server.Client
 			if (token.Bearer == null)
 				throw new InvalidOperationException("token.Bearer should not be null!");
 
-			return new ServerClient(ApiClientFactory.CreateApiClient(host, new ApiHeaders(productHeaderValue, token.Bearer), null, false), token);
+			var serverClient = new ServerClient(
+				ApiClientFactory.CreateApiClient(
+					host,
+					new ApiHeaders(
+						productHeaderValue,
+						token),
+					null,
+					false));
+			return serverClient;
 		}
 
 		/// <inheritdoc />
@@ -112,7 +120,16 @@ namespace Tgstation.Server.Client
 			TimeSpan? timeout = null,
 			CancellationToken cancellationToken = default)
 		{
-			using var api = ApiClientFactory.CreateApiClient(host, new ApiHeaders(productHeaderValue, "fake"), null, true);
+			await using var api = ApiClientFactory.CreateApiClient(
+				host,
+				new ApiHeaders(
+					productHeaderValue,
+					new TokenResponse
+					{
+						Bearer = "unused",
+					}),
+				null,
+				true);
 
 			if (requestLoggers != null)
 				foreach (var requestLogger in requestLoggers)
@@ -145,7 +162,7 @@ namespace Tgstation.Server.Client
 			requestLoggers ??= Enumerable.Empty<IRequestLogger>();
 
 			TokenResponse token;
-			using (var api = ApiClientFactory.CreateApiClient(host, loginHeaders, null, false))
+			await using (var api = ApiClientFactory.CreateApiClient(host, loginHeaders, null, false))
 			{
 				foreach (var requestLogger in requestLoggers)
 					api.AddRequestLogger(requestLogger);
@@ -155,14 +172,13 @@ namespace Tgstation.Server.Client
 				token = await api.Update<TokenResponse>(Routes.Root, cancellationToken).ConfigureAwait(false);
 			}
 
-			var apiHeaders = new ApiHeaders(productHeaderValue, token.Bearer!);
+			var apiHeaders = new ApiHeaders(productHeaderValue, token);
 			var client = new ServerClient(
 				ApiClientFactory.CreateApiClient(
 					host,
 					apiHeaders,
 					attemptLoginRefresh ? loginHeaders : null,
-					false),
-				token);
+					false));
 			if (timeout.HasValue)
 				client.Timeout = timeout.Value;
 

src/Tgstation.Server.Client/Tgstation.Server.Client.csproj

@@ -9,6 +9,13 @@
     <PackageReleaseNotes>$(TGS_NUGET_RELEASE_NOTES_CLIENT)</PackageReleaseNotes>
   </PropertyGroup>
 
+  <ItemGroup>
+    <!-- Usage: Connecting to SignalR hubs in API -->
+    <PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="8.0.0-rc.1.23421.29" />
+    <!-- Usage: Using target JSON serializer for API -->
+    <PackageReference Include="Microsoft.AspNetCore.SignalR.Protocols.NewtonsoftJson" Version="8.0.0-rc.1.23421.29" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\Tgstation.Server.Api\Tgstation.Server.Api.csproj" />
     <ProjectReference Include="..\Tgstation.Server.Common\Tgstation.Server.Common.csproj" /> <!-- Needed for explicit nuget versioning -->