Subsystem 1b-ii: live connection lifecycle, failover & #info status

Directory.Packages.props

@@ -11,6 +11,7 @@
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.9" />
     <PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.9" />
     <PackageVersion Include="Persistord.Core" Version="1.0.0-beta.1" />
+    <PackageVersion Include="RustPlusApi" Version="2.0.0-beta.1" />
     <PackageVersion Include="RustPlusApi.Fcm" Version="2.0.0-beta.1" />
   </ItemGroup>
   <ItemGroup>

RustPlusBot.slnx

@@ -4,12 +4,14 @@
     <Project Path="src/RustPlusBot.Discord/RustPlusBot.Discord.csproj" />
     <Project Path="src/RustPlusBot.Domain/RustPlusBot.Domain.csproj" />
     <Project Path="src/RustPlusBot.Host/RustPlusBot.Host.csproj" />
+    <Project Path="src/RustPlusBot.Features.Connections/RustPlusBot.Features.Connections.csproj" />
     <Project Path="src/RustPlusBot.Features.Pairing/RustPlusBot.Features.Pairing.csproj" />
     <Project Path="src/RustPlusBot.Features.Workspace/RustPlusBot.Features.Workspace.csproj" />
     <Project Path="src/RustPlusBot.Persistence/RustPlusBot.Persistence.csproj" />
   </Folder>
   <Folder Name="/tests/">
     <Project Path="tests/RustPlusBot.Abstractions.Tests/RustPlusBot.Abstractions.Tests.csproj" />
+    <Project Path="tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj" />
     <Project Path="tests/RustPlusBot.Features.Pairing.Tests/RustPlusBot.Features.Pairing.Tests.csproj" />
     <Project Path="tests/RustPlusBot.Features.Workspace.Tests/RustPlusBot.Features.Workspace.Tests.csproj" />
     <Project Path="tests/RustPlusBot.Persistence.Tests/RustPlusBot.Persistence.Tests.csproj" />

src/RustPlusBot.Abstractions/Events/ConnectionStatusChangedEvent.cs

@@ -0,0 +1,6 @@
+namespace RustPlusBot.Abstractions.Events;
+
+/// <summary>Published when a server's live-connection state changes, so #info can re-render.</summary>
+/// <param name="GuildId">The owning guild snowflake.</param>
+/// <param name="ServerId">The server whose connection state changed.</param>
+public sealed record ConnectionStatusChangedEvent(ulong GuildId, Guid ServerId);

src/RustPlusBot.Discord/DiscordServiceCollectionExtensions.cs

@@ -2,6 +2,7 @@
 using Discord.Interactions;
 using Discord.WebSocket;
 using Microsoft.Extensions.DependencyInjection;
+using RustPlusBot.Discord.Notifications;
 
 namespace RustPlusBot.Discord;
 
@@ -28,6 +29,7 @@ public static IServiceCollection AddDiscordBot(this IServiceCollection services)
             {
                 DefaultRunMode = RunMode.Async
             }));
+        services.AddSingleton<IUserDmSender, DiscordUserDmSender>();
         services.AddHostedService<DiscordBotService>();
 
         return services;

src/RustPlusBot.Discord/Notifications/DiscordUserDmSender.cs

@@ -0,0 +1,44 @@
+using System.Diagnostics.CodeAnalysis;
+using Discord.WebSocket;
+using Microsoft.Extensions.Logging;
+
+namespace RustPlusBot.Discord.Notifications;
+
+/// <summary>Discord-backed <see cref="IUserDmSender"/>.</summary>
+/// <param name="client">The socket client.</param>
+/// <param name="logger">The logger.</param>
+[SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes",
+    Justification = "Instantiated by the DI container via IUserDmSender registration.")]
+internal sealed partial class DiscordUserDmSender(DiscordSocketClient client, ILogger<DiscordUserDmSender> logger)
+    : IUserDmSender
+{
+    /// <inheritdoc />
+    public async Task SendAsync(ulong userId, string message, CancellationToken cancellationToken = default)
+    {
+        try
+        {
+            // Discord.Net uses RequestOptions, not CancellationToken, for per-call cancellation, so the token is not forwarded.
+            var user = await client.GetUserAsync(userId).ConfigureAwait(false);
+            if (user is null)
+            {
+                LogUserNotFound(logger, userId);
+                return;
+            }
+
+            var dmChannel = await user.CreateDMChannelAsync().ConfigureAwait(false);
+            await dmChannel.SendMessageAsync(message).ConfigureAwait(false);
+        }
+#pragma warning disable CA1031 // Broad catch is intentional: a closed DM (or any send failure) must not break callers.
+        catch (Exception ex)
+#pragma warning restore CA1031
+        {
+            LogDmFailed(logger, ex, userId);
+        }
+    }
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "Cannot DM user {UserId}: user not found.")]
+    private static partial void LogUserNotFound(ILogger logger, ulong userId);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "Failed to DM user {UserId}.")]
+    private static partial void LogDmFailed(ILogger logger, Exception ex, ulong userId);
+}

src/RustPlusBot.Discord/Notifications/IUserDmSender.cs

@@ -0,0 +1,12 @@
+namespace RustPlusBot.Discord.Notifications;
+
+/// <summary>Sends a direct message to a Discord user. A closed DM (or any send failure) is swallowed and logged.</summary>
+public interface IUserDmSender
+{
+    /// <summary>Best-effort DM to <paramref name="userId"/>; never throws.</summary>
+    /// <param name="userId">The Discord user snowflake.</param>
+    /// <param name="message">The message text.</param>
+    /// <param name="cancellationToken">A cancellation token.</param>
+    /// <returns>A task that completes when the send was attempted.</returns>
+    Task SendAsync(ulong userId, string message, CancellationToken cancellationToken = default);
+}

src/RustPlusBot.Domain/Connections/ConnectionState.cs

@@ -1,6 +1,6 @@
 namespace RustPlusBot.Domain.Connections;
 
-/// <summary>Persisted last-known connection state per server, so the active identity survives restarts.</summary>
+/// <summary>Persisted last-known connection state per server, so the active identity and status survive restarts.</summary>
 public sealed class ConnectionState
 {
     /// <summary>The server this state belongs to (primary key, one row per server).</summary>
@@ -12,8 +12,11 @@ public sealed class ConnectionState
     /// <summary>The credential currently selected as active, if any.</summary>
     public Guid? ActiveCredentialId { get; set; }
 
-    /// <summary>Whether the connection was healthy at last check.</summary>
-    public bool IsHealthy { get; set; }
+    /// <summary>The live-connection status.</summary>
+    public ConnectionStatus Status { get; set; }
+
+    /// <summary>Last heartbeat player count, or null if unknown.</summary>
+    public int? PlayerCount { get; set; }
 
     /// <summary>When the state was last updated (UTC).</summary>
     public DateTimeOffset UpdatedAt { get; set; }

src/RustPlusBot.Domain/Connections/ConnectionStatus.cs

@@ -0,0 +1,17 @@
+namespace RustPlusBot.Domain.Connections;
+
+/// <summary>Live-connection lifecycle state for a server's socket.</summary>
+public enum ConnectionStatus
+{
+    /// <summary>Attempting to connect (initial, after a swap, or after a drop).</summary>
+    Connecting = 0,
+
+    /// <summary>Socket up and the last heartbeat was healthy.</summary>
+    Connected = 1,
+
+    /// <summary>A valid active credential exists but the server is not answering; retrying with backoff.</summary>
+    Unreachable = 2,
+
+    /// <summary>No eligible credential (pool empty or every credential is Invalid).</summary>
+    NoCredentials = 3,
+}

src/RustPlusBot.Features.Connections/ConnectionOptions.cs

@@ -0,0 +1,20 @@
+namespace RustPlusBot.Features.Connections;
+
+/// <summary>Connection feature configuration, bound from the "Connections" config section.</summary>
+public sealed class ConnectionOptions
+{
+    /// <summary>How long to wait for a socket to connect before treating the server as unreachable.</summary>
+    public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(30);
+
+    /// <summary>First delay before retrying after an unreachable result.</summary>
+    public TimeSpan InitialRetryDelay { get; set; } = TimeSpan.FromSeconds(5);
+
+    /// <summary>Cap on the exponential backoff between reconnect attempts.</summary>
+    public TimeSpan MaxRetryDelay { get; set; } = TimeSpan.FromMinutes(5);
+
+    /// <summary>How often to send a heartbeat on a connected socket.</summary>
+    public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromSeconds(60);
+
+    /// <summary>How long a single heartbeat may take before the socket is considered unreachable.</summary>
+    public TimeSpan HeartbeatTimeout { get; set; } = TimeSpan.FromSeconds(10);
+}

src/RustPlusBot.Features.Connections/ConnectionServiceCollectionExtensions.cs

@@ -0,0 +1,29 @@
+using Microsoft.Extensions.DependencyInjection;
+using RustPlusBot.Discord;
+using RustPlusBot.Features.Connections.Hosting;
+using RustPlusBot.Features.Connections.Listening;
+using RustPlusBot.Features.Connections.Supervisor;
+
+namespace RustPlusBot.Features.Connections;
+
+/// <summary>DI registration for the live-connection feature.</summary>
+public static class ConnectionServiceCollectionExtensions
+{
+    /// <summary>Registers the socket source, supervisor, module seam, and hosted service.</summary>
+    /// <param name="services">The service collection to add to.</param>
+    /// <returns>The same service collection, for chaining.</returns>
+    public static IServiceCollection AddConnections(this IServiceCollection services)
+    {
+        ArgumentNullException.ThrowIfNull(services);
+
+        services.AddSingleton<IRustSocketSource, RustPlusSocketSource>();
+        services.AddSingleton<IConnectionSupervisor, ConnectionSupervisor>();
+
+        // Contribute this assembly's interaction modules to the Discord layer.
+        services.AddSingleton(new InteractionModuleAssembly(typeof(ConnectionServiceCollectionExtensions).Assembly));
+
+        services.AddHostedService<ConnectionHostedService>();
+
+        return services;
+    }
+}

src/RustPlusBot.Features.Connections/Hosting/ConnectionHostedService.cs

@@ -0,0 +1,81 @@
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using RustPlusBot.Abstractions.Events;
+using RustPlusBot.Features.Connections.Supervisor;
+
+namespace RustPlusBot.Features.Connections.Hosting;
+
+/// <summary>Drives the connection supervisor: start all on startup, react to server registration, stop on shutdown.</summary>
+/// <param name="supervisor">The connection supervisor.</param>
+/// <param name="eventBus">The in-process event bus.</param>
+/// <param name="logger">The logger.</param>
+internal sealed partial class ConnectionHostedService(
+    IConnectionSupervisor supervisor,
+    IEventBus eventBus,
+    ILogger<ConnectionHostedService> logger) : IHostedService, IDisposable
+{
+    private readonly CancellationTokenSource _cts = new();
+    private Task? _eventLoop;
+
+    /// <inheritdoc />
+    public void Dispose() => _cts.Dispose();
+
+    /// <inheritdoc />
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        _eventLoop = Task.Run(() => RunAsync(_cts.Token), CancellationToken.None);
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc />
+    public async Task StopAsync(CancellationToken cancellationToken)
+    {
+        await _cts.CancelAsync().ConfigureAwait(false);
+        if (_eventLoop is not null)
+        {
+            try
+            {
+#pragma warning disable VSTHRD003 // Suppress: this is our own loop task, joined on stop.
+                await _eventLoop.ConfigureAwait(false);
+#pragma warning restore VSTHRD003
+            }
+            catch (OperationCanceledException)
+            {
+                // Expected on shutdown.
+            }
+        }
+
+        await supervisor.StopAllAsync().ConfigureAwait(false);
+    }
+
+    private async Task RunAsync(CancellationToken cancellationToken)
+    {
+        // Subscription is registered when this loop first awaits the bus, after StartAllAsync completes.
+        // The in-process bus does not replay, so events published before this point are not delivered. Safe:
+        // the server-registered event is only raised by the FCM pairing flow at runtime, long after startup.
+        try
+        {
+            await supervisor.StartAllAsync(cancellationToken).ConfigureAwait(false);
+
+            await foreach (var registered in eventBus.SubscribeAsync<ServerRegisteredEvent>(cancellationToken)
+                               .ConfigureAwait(false))
+            {
+                await supervisor.EnsureConnectionAsync(registered.GuildId, registered.ServerId, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            // Shutting down.
+        }
+#pragma warning disable CA1031 // Broad catch is intentional: a faulting consumer must not crash the host.
+        catch (Exception ex)
+        {
+            LogLoopFaulted(logger, ex);
+        }
+#pragma warning restore CA1031
+    }
+
+    [LoggerMessage(Level = LogLevel.Error, Message = "Connection hosted-service loop faulted.")]
+    private static partial void LogLoopFaulted(ILogger logger, Exception exception);
+}

src/RustPlusBot.Features.Connections/Listening/HeartbeatResult.cs

@@ -0,0 +1,32 @@
+namespace RustPlusBot.Features.Connections.Listening;
+
+/// <summary>Classification of a heartbeat probe.</summary>
+internal enum HeartbeatKind
+{
+    /// <summary>Server answered; <see cref="HeartbeatResult.PlayerCount"/> is valid.</summary>
+    Ok = 0,
+
+    /// <summary>Server did not answer in time.</summary>
+    Unreachable = 1,
+
+    /// <summary>Server rejected the request as unauthorized (token died mid-session).</summary>
+    AuthRejected = 2,
+}
+
+/// <summary>The outcome of a heartbeat probe.</summary>
+/// <param name="Kind">The classification.</param>
+/// <param name="PlayerCount">Players online (only meaningful when <see cref="HeartbeatKind.Ok"/>).</param>
+/// <remarks><see cref="Ok"/> is a method because it carries a player count; <see cref="Unreachable"/>
+/// and <see cref="AuthRejected"/> are constants.</remarks>
+internal readonly record struct HeartbeatResult(HeartbeatKind Kind, int PlayerCount)
+{
+    /// <summary>An unreachable heartbeat.</summary>
+    public static HeartbeatResult Unreachable { get; } = new(HeartbeatKind.Unreachable, 0);
+
+    /// <summary>An auth-rejected heartbeat.</summary>
+    public static HeartbeatResult AuthRejected { get; } = new(HeartbeatKind.AuthRejected, 0);
+
+    /// <summary>A healthy heartbeat carrying the player count.</summary>
+    /// <param name="playerCount">The number of players currently online.</param>
+    public static HeartbeatResult Ok(int playerCount) => new(HeartbeatKind.Ok, playerCount);
+}

src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs

@@ -0,0 +1,17 @@
+namespace RustPlusBot.Features.Connections.Listening;
+
+/// <summary>One live Rust+ socket to a single server, driven by one player credential.</summary>
+internal interface IRustServerConnection : IAsyncDisposable
+{
+    /// <summary>Connects within <paramref name="timeout"/>.</summary>
+    /// <param name="timeout">How long to wait for the connection.</param>
+    /// <param name="cancellationToken">A cancellation token.</param>
+    /// <returns>The connect outcome.</returns>
+    Task<SocketConnectOutcome> ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken);
+
+    /// <summary>Sends a lightweight info request as a heartbeat (and auth probe), within <paramref name="timeout"/>.</summary>
+    /// <param name="timeout">How long to wait for the response.</param>
+    /// <param name="cancellationToken">A cancellation token.</param>
+    /// <returns>The heartbeat result.</returns>
+    Task<HeartbeatResult> GetInfoAsync(TimeSpan timeout, CancellationToken cancellationToken);
+}

src/RustPlusBot.Features.Connections/Listening/IRustSocketSource.cs

@@ -0,0 +1,13 @@
+namespace RustPlusBot.Features.Connections.Listening;
+
+/// <summary>Creates <see cref="IRustServerConnection"/>s (the real RustPlusApi adapter in production, a fake in tests).</summary>
+internal interface IRustSocketSource
+{
+    /// <summary>Creates a connection to <paramref name="ip"/>:<paramref name="port"/> as the given player.</summary>
+    /// <param name="ip">Server host.</param>
+    /// <param name="port">Server Rust+ port.</param>
+    /// <param name="steamId">The player's Steam64 id.</param>
+    /// <param name="playerToken">The player's Rust+ token (numeric, as a string).</param>
+    /// <returns>A new connection.</returns>
+    IRustServerConnection Create(string ip, int port, ulong steamId, string playerToken);
+}