Compare commits

..
7 changed files with 87 additions and 22 deletions
+13
View File
@@ -122,6 +122,19 @@ namespace Notesnook.API.Hubs
await base.OnConnectedAsync();
}
public override async Task OnDisconnectedAsync(Exception? exception)
{
if (exception != null)
{
Logger.LogWarning(exception, "Connection {ConnectionId} disconnected with error (server-side drop)", Context.ConnectionId);
}
else
{
Logger.LogInformation("Connection {ConnectionId} disconnected cleanly (client-initiated)", Context.ConnectionId);
}
await base.OnDisconnectedAsync(exception);
}
public async Task<int> PushItems(string deviceId, SyncTransferItemV2 pushItem)
{
+1
View File
@@ -9,6 +9,7 @@
<ItemGroup>
<PackageReference Include="AngleSharp" Version="1.3.0" />
<PackageReference Include="AspNetCore.HealthChecks.Aws.S3" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.Redis" Version="9.0.0" />
<PackageReference Include="AWSSDK.Core" Version="3.7.304.31" />
<PackageReference Include="DotNetEnv" Version="2.3.0" />
<PackageReference Include="IdentityModel.AspNetCore.OAuth2Introspection" Version="6.2.0" />
+15 -1
View File
@@ -25,6 +25,7 @@ using System.Text;
using System.Text.Encodings.Web;
using System.Threading.Tasks;
using Amazon.Runtime;
using StackExchange.Redis;
using IdentityModel.AspNetCore.OAuth2Introspection;
using Microsoft.AspNetCore.Authentication;
using Microsoft.AspNetCore.Authentication.JwtBearer;
@@ -219,7 +220,20 @@ namespace Notesnook.API
}).AddMessagePackProtocol().AddJsonProtocol();
if (!string.IsNullOrEmpty(Constants.SIGNALR_REDIS_CONNECTION_STRING))
signalR.AddStackExchangeRedis(Constants.SIGNALR_REDIS_CONNECTION_STRING);
{
services.AddHealthChecks()
.AddRedis(Constants.SIGNALR_REDIS_CONNECTION_STRING, tags: ["ready"]);
signalR.AddStackExchangeRedis(options =>
{
options.Configuration = ConfigurationOptions.Parse(Constants.SIGNALR_REDIS_CONNECTION_STRING);
options.Configuration.AbortOnConnectFail = false;
options.Configuration.ConnectRetry = 5;
options.Configuration.ReconnectRetryPolicy = new ExponentialRetry(5000, 30000);
options.Configuration.KeepAlive = 60;
options.Configuration.ConnectTimeout = 5000;
options.Configuration.SyncTimeout = 5000;
});
}
services.AddResponseCompression(options =>
{
@@ -190,8 +190,8 @@ namespace Streetwriters.Identity.Controllers
var client = Clients.FindClientById(form.ClientId);
if (client == null) return BadRequest("Invalid client_id.");
var user = await UserManager.FindByEmailAsync(form.Email) ?? throw new Exception("User not found.");
if (!await UserService.IsUserValidAsync(UserManager, user, form.ClientId)) return Ok();
var user = await UserManager.FindByEmailAsync(form.Email);
if (user == null || !await UserService.IsUserValidAsync(UserManager, user, form.ClientId)) return Ok();
var code = await UserManager.GenerateUserTokenAsync(user, TokenOptions.DefaultProvider, "ResetPassword");
var callbackUrl = UrlExtensions.TokenLink(user.Id.ToString(), code, client.Id, TokenType.RESET_PASSWORD);
+31 -10
View File
@@ -18,28 +18,49 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
using System.Linq;
using System;
using System.Threading;
using System.Threading.Tasks;
using Lib.AspNetCore.ServerSentEvents;
using System.Security.Claims;
using System.Collections.Generic;
namespace Streetwriters.Messenger.Helpers
{
public class SSEHelper
{
public static async Task SendEventToUserAsync(string data, IServerSentEventsService sseService, string userId, string? originTokenId = null)
public static async Task SendEventToUserAsync(string data, IServerSentEventsService sseService, string userId, string? originTokenId = null, CancellationToken cancellationToken = default)
{
var clients = sseService.GetClients().Where(c => c.User.FindFirstValue("sub") == userId);
foreach (var client in clients)
{
if (originTokenId != null && client.User.FindFirstValue("jti") == originTokenId) continue;
if (!client.IsConnected) continue;
await client.SendEventAsync(data);
}
var clients = sseService.GetClients()
.Where(c => c.User?.FindFirstValue("sub") == userId)
.Where(c => originTokenId == null || c.User?.FindFirstValue("jti") != originTokenId);
await SendEventToClientsAsync(clients, data, cancellationToken);
}
public static async Task SendEventToAllUsersAsync(string data, IServerSentEventsService sseService)
public static async Task SendEventToAllUsersAsync(string data, IServerSentEventsService sseService, CancellationToken cancellationToken = default)
{
await sseService.SendEventAsync(data);
await SendEventToClientsAsync(sseService.GetClients(), data, cancellationToken);
}
private static async Task SendEventToClientsAsync(IEnumerable<IServerSentEventsClient> clients, string data, CancellationToken cancellationToken)
{
foreach (var client in clients)
{
if (!client.IsConnected) continue;
try
{
await client.SendEventAsync(data, cancellationToken);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
break;
}
catch
{
}
}
}
}
}
@@ -21,6 +21,7 @@ using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Lib.AspNetCore.ServerSentEvents;
using Streetwriters.Messenger.Helpers;
using System.Text.Json;
@@ -33,12 +34,14 @@ namespace Streetwriters.Messenger.Services
private const string HEARTBEAT_MESSAGE_FORMAT = "Streetwriters Heartbeat ({0} UTC)";
private readonly IServerSentEventsService _serverSentEventsService;
private readonly ILogger<HeartbeatService> _logger;
#endregion
#region Constructor
public HeartbeatService(IServerSentEventsService serverSentEventsService)
public HeartbeatService(IServerSentEventsService serverSentEventsService, ILogger<HeartbeatService> logger)
{
_serverSentEventsService = serverSentEventsService;
_logger = logger;
}
#endregion
@@ -47,15 +50,28 @@ namespace Streetwriters.Messenger.Services
{
while (!stoppingToken.IsCancellationRequested)
{
var message = JsonSerializer.Serialize(new
try
{
type = "heartbeat",
data = JsonSerializer.Serialize(new
var message = JsonSerializer.Serialize(new
{
t = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
})
});
await SSEHelper.SendEventToAllUsersAsync(message, _serverSentEventsService);
type = "heartbeat",
data = JsonSerializer.Serialize(new
{
t = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
})
});
await SSEHelper.SendEventToAllUsersAsync(message, _serverSentEventsService, stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
break;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to send SSE heartbeat to one or more clients.");
}
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
}
}
@@ -8,7 +8,7 @@
<ItemGroup>
<PackageReference Include="DotNetEnv" Version="2.3.0" />
<PackageReference Include="Lib.AspNetCore.ServerSentEvents" Version="6.0.0" />
<PackageReference Include="Lib.AspNetCore.ServerSentEvents" Version="9.1.0" />
<PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="5.0.0"
NoWarn="NU1605" />
<PackageReference Include="Microsoft.AspNetCore.Authentication.OpenIdConnect" Version="5.0.0"