feat: making it work even better...i abandon writing tests first...😅

This commit is contained in:
Stevan Freeborn
2025-05-15 16:07:22 -05:00
parent 23a8bbb3a6
commit 3a2129f860
9 changed files with 183 additions and 40 deletions
+1 -1
View File
@@ -65,7 +65,7 @@ dotnet_style_prefer_auto_properties = true:suggestion
dotnet_style_prefer_collection_expression = when_types_loosely_match:suggestion dotnet_style_prefer_collection_expression = when_types_loosely_match:suggestion
dotnet_style_prefer_compound_assignment = true:suggestion dotnet_style_prefer_compound_assignment = true:suggestion
dotnet_style_prefer_conditional_expression_over_assignment = true:suggestion dotnet_style_prefer_conditional_expression_over_assignment = true:suggestion
dotnet_style_prefer_conditional_expression_over_return = true:suggestion dotnet_style_prefer_conditional_expression_over_return = false:silent
dotnet_style_prefer_foreach_explicit_cast_in_source = when_strongly_typed:suggestion dotnet_style_prefer_foreach_explicit_cast_in_source = when_strongly_typed:suggestion
dotnet_style_prefer_inferred_anonymous_type_member_names = true:suggestion dotnet_style_prefer_inferred_anonymous_type_member_names = true:suggestion
dotnet_style_prefer_inferred_tuple_names = true:suggestion dotnet_style_prefer_inferred_tuple_names = true:suggestion
@@ -0,0 +1,21 @@
namespace StevesBot.Worker.Discord;
internal static class DiscordCloseCodes
{
private const int AuthenticationFailed = 4004;
private const int InvalidShard = 4010;
private const int ShardingRequired = 4011;
private const int InvalidAPIVersion = 4012;
private const int InvalidIntents = 4013;
private const int DisallowedIntents = 4014;
public static bool IsReconnectable(int? closeCode)
{
return closeCode is not AuthenticationFailed
and not InvalidShard
and not ShardingRequired
and not InvalidAPIVersion
and not InvalidIntents
and not DisallowedIntents;
}
}
@@ -4,6 +4,11 @@ namespace StevesBot.Worker.Discord;
internal sealed class DiscordGatewayClient : IDiscordGatewayClient internal sealed class DiscordGatewayClient : IDiscordGatewayClient
{ {
private const string VersionKey = "v";
private const string VersionValue = "10";
private const string EncodingKey = "encoding";
private const string EncodingValue = "json";
private readonly DiscordClientOptions _options; private readonly DiscordClientOptions _options;
private readonly IWebSocketFactory _webSocketFactory; private readonly IWebSocketFactory _webSocketFactory;
private readonly ILogger<DiscordGatewayClient> _logger; private readonly ILogger<DiscordGatewayClient> _logger;
@@ -23,6 +28,7 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
private string _gatewayUrl = string.Empty; private string _gatewayUrl = string.Empty;
private IWebSocket? _webSocket; private IWebSocket? _webSocket;
private int? _lastSequence; private int? _lastSequence;
private int? _lastDispatchSequence;
private int _heartbeatInterval; private int _heartbeatInterval;
private DateTimeOffset _timeLastHeartbeatSent = DateTimeOffset.MinValue; private DateTimeOffset _timeLastHeartbeatSent = DateTimeOffset.MinValue;
private DateTimeOffset _timeLastHeartbeatAcknowledged = DateTimeOffset.MinValue; private DateTimeOffset _timeLastHeartbeatAcknowledged = DateTimeOffset.MinValue;
@@ -94,11 +100,6 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
while (_linkedReceiveMessageCts?.IsCancellationRequested is false) while (_linkedReceiveMessageCts?.IsCancellationRequested is false)
{ {
// websocket message might be larger than the
// size of the buffer so we need to loop until
// we receive the end of the message and
// write the data we receive on each iteration
// to the memory stream
using var memoryStream = new MemoryStream(); using var memoryStream = new MemoryStream();
WebSocketReceiveResult result; WebSocketReceiveResult result;
@@ -108,16 +109,23 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
if (result.MessageType is WebSocketMessageType.Close) if (result.MessageType is WebSocketMessageType.Close)
{ {
// TODO: Handle all possible close codes if (DiscordCloseCodes.IsReconnectable((int?)result.CloseStatus) is false)
// reconnect accordingly to documentation {
var canResume = result.CloseStatus is not WebSocketCloseStatus.NormalClosure or WebSocketCloseStatus.EndpointUnavailable _logger.LogCritical("Received close status indicating we should not attempt reconnection: {CloseStatus}", result.CloseStatus);
&& string.IsNullOrWhiteSpace(_resumeGatewayUrl) is false await DisconnectAsync(cancellationToken);
&& string.IsNullOrWhiteSpace(_sessionId) is false; return;
}
var canResume = IsResumableCloseCode(result.CloseStatus);
var closeStatus = canResume ? WebSocketCloseStatus.Empty : WebSocketCloseStatus.NormalClosure;
await SetCanResumeAsync(canResume, _linkedReceiveMessageCts.Token); await SetCanResumeAsync(canResume, _linkedReceiveMessageCts.Token);
await CloseIfOpenAsync(result.CloseStatus ?? WebSocketCloseStatus.NormalClosure, result.CloseStatusDescription, _linkedReceiveMessageCts.Token); await CloseIfOpenAsync(closeStatus, result.CloseStatusDescription, _linkedReceiveMessageCts.Token);
_logger.LogInformation("Reconnecting because of close message: {CloseStatus} - {CloseStatusDescription}", result.CloseStatus, result.CloseStatusDescription);
_logger.LogWarning("Reconnecting because of close message: {CloseStatus} - {CloseStatusDescription}", result.CloseStatus, result.CloseStatusDescription);
await ReconnectAsync(cancellationToken); await ReconnectAsync(cancellationToken);
return; return;
} }
@@ -165,7 +173,9 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
); );
await SetCanResumeAsync(false, newReceiveLinkedCts.Token); await SetCanResumeAsync(false, newReceiveLinkedCts.Token);
_logger.LogInformation("Reconnecting because of error in receive message task");
_logger.LogWarning("Reconnecting because of error in receive message task");
await ReconnectAsync(cancellationToken); await ReconnectAsync(cancellationToken);
} }
}, newReceiveLinkedCts.Token); }, newReceiveLinkedCts.Token);
@@ -182,16 +192,77 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
await SetHeartbeatIntervalAsync(he.Data.HeartbeatInterval, cancellationToken); await SetHeartbeatIntervalAsync(he.Data.HeartbeatInterval, cancellationToken);
await StartHeartbeatAsync(cancellationToken); await StartHeartbeatAsync(cancellationToken);
await IdentifyAsync(cancellationToken); await IdentifyAsync(cancellationToken);
_logger.LogInformation("Hello event received. Heartbeat interval: {Interval}", he.Data.HeartbeatInterval); _logger.LogInformation("Hello event received");
break; break;
case HeartbeatAckDiscordEvent: case HeartbeatAckDiscordEvent:
await SetHeartbeatAcknowledgedAsync(_timeProvider.GetUtcNow(), cancellationToken); await SetHeartbeatAcknowledgedAsync(_timeProvider.GetUtcNow(), cancellationToken);
_logger.LogInformation("Heartbeat acknowledged"); _logger.LogInformation("Heartbeat acknowledged");
break; break;
case ReadyDiscordEvent re: case HeartbeatDiscordEvent:
await SetSessionIdAsync(re.Data.SessionId, cancellationToken); _logger.LogInformation("Heartbeat request received");
await SetResumeGatewayUrlAsync(re.Data.ResumeGatewayUrl, cancellationToken); await SendHeartbeatAsync(cancellationToken);
_logger.LogInformation("Ready event received"); break;
case DispatchDiscordEvent de:
await SetDispatchSequenceAsync(e.Sequence, cancellationToken);
switch (de)
{
case ReadyDiscordEvent re:
await SetSessionIdAsync(re.Data.SessionId, cancellationToken);
await SetResumeGatewayUrlAsync(re.Data.ResumeGatewayUrl, cancellationToken);
_logger.LogInformation("Ready event received");
break;
default:
_logger.LogInformation("Received dispatch event: {Event}", de.Type ?? "Unknown");
break;
}
break;
case ReconnectDiscordEvent:
_logger.LogInformation("Reconnect event received");
_logger.LogWarning("Closing connection without invalidating session.");
await SetCanResumeAsync(true, cancellationToken);
await CloseIfOpenAsync(
WebSocketCloseStatus.Empty,
"Reconnect event received",
cancellationToken
);
_logger.LogWarning("Reconnecting because of reconnect event");
await ReconnectAsync(cancellationToken);
break;
case InvalidSessionDiscordEvent ise:
_logger.LogInformation("Invalid session event received");
if (ise.Data)
{
_logger.LogWarning("Session is resumable. Closing connection without invalidating session.");
await SetCanResumeAsync(true, cancellationToken);
await CloseIfOpenAsync(
WebSocketCloseStatus.Empty,
"Invalid session event received",
cancellationToken
);
}
else
{
_logger.LogWarning("Session is not resumable. Closing connection and invalidating session.");
await SetCanResumeAsync(false, cancellationToken);
await CloseIfOpenAsync(
WebSocketCloseStatus.NormalClosure,
"Invalid session event received",
cancellationToken
);
}
await ReconnectAsync(cancellationToken);
break; break;
default: default:
_logger.LogInformation("Received event: {Event}", e.GetType().Name); _logger.LogInformation("Received event: {Event}", e.GetType().Name);
@@ -237,7 +308,9 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
} }
_logger.LogWarning("Heartbeat not acknowledged. Reconnecting."); _logger.LogWarning("Heartbeat not acknowledged. Reconnecting.");
await ReconnectAsync(cancellationToken); await ReconnectAsync(cancellationToken);
break; break;
} }
@@ -263,7 +336,8 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
newLinkedCts.Token newLinkedCts.Token
); );
_logger.LogInformation("Reconnecting because of error in heartbeat task"); _logger.LogWarning("Reconnecting because of error in heartbeat task");
await ReconnectAsync(cancellationToken); await ReconnectAsync(cancellationToken);
} }
} }
@@ -296,7 +370,7 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
using (await _lock.LockAsync(cancellationToken)) using (await _lock.LockAsync(cancellationToken))
{ {
if (_lastSequence is null) if (_lastDispatchSequence is null)
{ {
throw new DiscordGatewayClientException("Cannot resume without a sequence number."); throw new DiscordGatewayClientException("Cannot resume without a sequence number.");
} }
@@ -304,7 +378,7 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
resume = new ResumeDiscordEvent( resume = new ResumeDiscordEvent(
_options.AppToken, _options.AppToken,
_sessionId, _sessionId,
_lastSequence.Value _lastDispatchSequence.Value
); );
} }
@@ -360,6 +434,21 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
} }
} }
private bool IsResumableCloseCode(WebSocketCloseStatus? closeStatus)
{
if (string.IsNullOrWhiteSpace(_resumeGatewayUrl) || string.IsNullOrWhiteSpace(_sessionId))
{
return false;
}
if (closeStatus is WebSocketCloseStatus.NormalClosure or WebSocketCloseStatus.EndpointUnavailable)
{
return false;
}
return true;
}
private bool IsWebSocketOpen() private bool IsWebSocketOpen()
{ {
return _webSocket?.State is WebSocketState.Open; return _webSocket?.State is WebSocketState.Open;
@@ -486,22 +575,19 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
return result; return result;
} }
private async Task ConnectWithGatewayUrlAsync(CancellationToken cancellationToken) private Task ConnectWithGatewayUrlAsync(CancellationToken cancellationToken)
{ {
if (_webSocket is null) var uri = BuildUri(_gatewayUrl);
{ return ConnectWithUriAsync(uri, cancellationToken);
throw new DiscordGatewayClientException("WebSocket is not set. Cannot connect.");
}
if (_webSocket.State is WebSocketState.Open)
{
throw new DiscordGatewayClientException("WebSocket is already open. Cannot connect.");
}
await _webSocket.ConnectAsync(new Uri(_gatewayUrl), cancellationToken);
} }
private async Task ConnectWithResumeUrlAsync(CancellationToken cancellationToken) private Task ConnectWithResumeUrlAsync(CancellationToken cancellationToken)
{
var uri = BuildUri(_resumeGatewayUrl);
return ConnectWithUriAsync(uri, cancellationToken);
}
private async Task ConnectWithUriAsync(Uri uri, CancellationToken cancellationToken)
{ {
if (_webSocket is null) if (_webSocket is null)
{ {
@@ -513,7 +599,15 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
throw new DiscordGatewayClientException("WebSocket is already open. Cannot connect."); throw new DiscordGatewayClientException("WebSocket is already open. Cannot connect.");
} }
await _webSocket.ConnectAsync(new Uri(_resumeGatewayUrl), cancellationToken); await _webSocket.ConnectAsync(uri, cancellationToken);
}
private static Uri BuildUri(string url)
{
return new UriBuilder(url)
{
Query = $"{VersionKey}={VersionValue}&{EncodingKey}={EncodingValue}"
}.Uri;
} }
private async Task SetGatewayUrlAsync(string url, CancellationToken cancellationToken) private async Task SetGatewayUrlAsync(string url, CancellationToken cancellationToken)
@@ -522,6 +616,12 @@ internal sealed class DiscordGatewayClient : IDiscordGatewayClient
_gatewayUrl = url; _gatewayUrl = url;
} }
private async Task SetDispatchSequenceAsync(int? sequence, CancellationToken cancellationToken)
{
using var _ = await _lock.LockAsync(cancellationToken);
_lastDispatchSequence = sequence;
}
private async Task SetSequenceAsync(int? sequence, CancellationToken cancellationToken) private async Task SetSequenceAsync(int? sequence, CancellationToken cancellationToken)
{ {
using var _ = await _lock.LockAsync(cancellationToken); using var _ = await _lock.LockAsync(cancellationToken);
@@ -14,6 +14,9 @@ internal sealed class DiscordEventConverter : JsonConverter<DiscordEvent>
return op switch return op switch
{ {
DiscordOpCodes.Dispatch => DeserializeDispatchEvent(root, options, modifiedOptions), DiscordOpCodes.Dispatch => DeserializeDispatchEvent(root, options, modifiedOptions),
DiscordOpCodes.Heartbeat => JsonSerializer.Deserialize<HeartbeatDiscordEvent>(root.GetRawText(), options),
DiscordOpCodes.Reconnect => JsonSerializer.Deserialize<ReconnectDiscordEvent>(root.GetRawText(), options),
DiscordOpCodes.InvalidSession => JsonSerializer.Deserialize<InvalidSessionDiscordEvent>(root.GetRawText(), options),
DiscordOpCodes.Hello => JsonSerializer.Deserialize<HelloDiscordEvent>(root.GetRawText(), options), DiscordOpCodes.Hello => JsonSerializer.Deserialize<HelloDiscordEvent>(root.GetRawText(), options),
DiscordOpCodes.HeartbeatAck => JsonSerializer.Deserialize<HeartbeatAckDiscordEvent>(root.GetRawText(), options), DiscordOpCodes.HeartbeatAck => JsonSerializer.Deserialize<HeartbeatAckDiscordEvent>(root.GetRawText(), options),
_ => JsonSerializer.Deserialize<DiscordEvent>(root.GetRawText(), modifiedOptions), _ => JsonSerializer.Deserialize<DiscordEvent>(root.GetRawText(), modifiedOptions),
@@ -26,14 +29,14 @@ internal sealed class DiscordEventConverter : JsonConverter<DiscordEvent>
JsonSerializer.Serialize(writer, value, value.GetType(), modifiedOptions); JsonSerializer.Serialize(writer, value, value.GetType(), modifiedOptions);
} }
private static DiscordEvent? DeserializeDispatchEvent(JsonElement root, JsonSerializerOptions options, JsonSerializerOptions modifiedOptions) private static DispatchDiscordEvent? DeserializeDispatchEvent(JsonElement root, JsonSerializerOptions options, JsonSerializerOptions modifiedOptions)
{ {
var type = root.GetProperty("t").GetString(); var type = root.GetProperty("t").GetString();
return type switch return type switch
{ {
DiscordEventTypes.Ready => JsonSerializer.Deserialize<ReadyDiscordEvent>(root.GetRawText(), options), DiscordEventTypes.Ready => JsonSerializer.Deserialize<ReadyDiscordEvent>(root.GetRawText(), options),
_ => JsonSerializer.Deserialize<DiscordEvent>(root.GetRawText(), modifiedOptions), _ => JsonSerializer.Deserialize<DispatchDiscordEvent>(root.GetRawText(), modifiedOptions),
}; };
} }
@@ -6,6 +6,8 @@ internal static class DiscordOpCodes
public const int Heartbeat = 1; public const int Heartbeat = 1;
public const int Identify = 2; public const int Identify = 2;
public const int Resume = 6; public const int Resume = 6;
public const int Reconnect = 7;
public const int InvalidSession = 9;
public const int HeartbeatAck = 11; public const int HeartbeatAck = 11;
public const int Hello = 10; public const int Hello = 10;
} }
@@ -0,0 +1,5 @@
namespace StevesBot.Worker.Discord.Events;
internal record DispatchDiscordEvent : DiscordEvent
{
}
@@ -0,0 +1,7 @@
namespace StevesBot.Worker.Discord.Events;
internal sealed record InvalidSessionDiscordEvent : DiscordEvent
{
[JsonPropertyName("d")]
public new bool Data { get; init; }
}
@@ -1,6 +1,6 @@
namespace StevesBot.Worker.Discord.Events; namespace StevesBot.Worker.Discord.Events;
internal record ReadyDiscordEvent : DiscordEvent internal record ReadyDiscordEvent : DispatchDiscordEvent
{ {
[JsonPropertyName("d")] [JsonPropertyName("d")]
public new ReadyData Data { get; init; } = new ReadyData(); public new ReadyData Data { get; init; } = new ReadyData();
@@ -0,0 +1,5 @@
namespace StevesBot.Worker.Discord;
internal sealed record ReconnectDiscordEvent : DiscordEvent
{
}