websocket implementation with postgres backplane. Includes:

1) support for messages with ack
2) allows messages to be sent on user reconnect
3) updates order count when order received notification read.
4) deletes old messages
5) handles websocket reconnect
6) aws sticky sessions

Author: devinleighsmith

api/Controllers/NotificationsController.cs

@@ -1,4 +1,3 @@
-using System.Security.Claims;
 using System;
 using System.Linq;
 using System.Threading.Tasks;
@@ -7,12 +6,29 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Scv.Api.Helpers.Extensions;
+using Scv.Api.SignalR;
+using Scv.Db.Repositories;
 
 namespace Scv.Api.Hubs;
 
 [Authorize]
 public class NotificationsHub : Hub
 {
+    private readonly UserConnectionTracker _connectionTracker;
+
+    public NotificationsHub(UserConnectionTracker connectionTracker)
+    {
+        _connectionTracker = connectionTracker;
+    }
+
+    /// <summary>
+    /// Validates origin and user of this connection in order to safely send/receive notifications.
+    /// </summary>
+    /// <remarks>
+    /// Uses server-side auth to validate that this user is authorized to ack this message.
+    /// </remarks>
+    /// <returns>A task representing the asynchronous operation.</returns>
     public override Task OnConnectedAsync()
     {
         var httpContext = Context.GetHttpContext();
@@ -55,7 +71,7 @@ public override Task OnConnectedAsync()
             }
         }
 
-        var userId = Context.User?.FindFirstValue(ClaimTypes.NameIdentifier);
+        var userId = Context.User?.UserId();
         if (string.IsNullOrWhiteSpace(userId))
         {
             if (logger != null)
@@ -74,6 +90,120 @@ public override Task OnConnectedAsync()
 
         logger?.LogInformation("SignalR connection accepted for user {UserId}.", userId);
 
+        // Track users connected to this server, so that this server only attempts to send notifications to users with an active connection.
+        _connectionTracker.AddConnection(userId, Context.ConnectionId);
+
         return base.OnConnectedAsync();
     }
+
+    public override Task OnDisconnectedAsync(Exception exception)
+    {
+        var userId = Context.User?.UserId();
+        if (!string.IsNullOrWhiteSpace(userId))
+        {
+            _connectionTracker.RemoveConnection(userId, Context.ConnectionId);
+        }
+
+        return base.OnDisconnectedAsync(exception);
+    }
+
+    /// <summary>
+    /// Acknowledges a notification for the current user by marking it as read. Note that an ack just indicates the client JASPER app received the notification - the app is responsible for presenting 
+    /// notifications to users.
+    /// </summary>
+    /// <param name="ackGuid">The stable identifier of the notification to acknowledge.</param>
+    /// <remarks>
+    /// Uses server-side auth to validate that this user is authorized to ack this message.
+    /// </remarks>
+    /// <returns>A task representing the asynchronous operation.</returns>
+    public async Task AckNotification(Guid ackGuid)
+    {
+        var logger = Context.GetHttpContext()
+            ?.RequestServices.GetService<ILogger<NotificationsHub>>();
+        var userId = Context.User?.UserId();
+        if (string.IsNullOrWhiteSpace(userId))
+        {
+            logger?.LogWarning(
+                "SignalR ack failed due to missing user id. AckGuid={AckGuid}",
+                ackGuid);
+            return;
+        }
+
+        var repository = Context.GetHttpContext()
+            ?.RequestServices.GetService<INotificationRepository>();
+        if (repository == null)
+        {
+            logger?.LogWarning(
+                "SignalR ack failed due to missing notification repository. AckGuid={AckGuid} UserId={UserId}",
+                ackGuid,
+                userId);
+            return;
+        }
+
+        var matches = await repository.FindAsync(message => message.AckGuid == ackGuid);
+        var message = matches.FirstOrDefault();
+        if (message == null)
+        {
+            logger?.LogWarning(
+                "SignalR ack failed to find message. AckGuid={AckGuid} UserId={UserId}",
+                ackGuid,
+                userId);
+            return;
+        }
+
+        if (!IsAckAuthorized(message.UserId, userId, out var reason))
+        {
+            logger?.LogWarning(
+                "SignalR ack rejected. AckGuid={AckGuid} UserId={UserId} Reason={Reason}",
+                ackGuid,
+                userId,
+                reason);
+            return;
+        }
+
+        if (!message.AckRequired)
+        {
+            logger?.LogInformation(
+                "SignalR ack skipped because ack is not required. AckGuid={AckGuid} UserId={UserId}",
+                ackGuid,
+                userId);
+            return;
+        }
+
+        if (message.AckedAt.HasValue)
+        {
+            logger?.LogInformation(
+                "SignalR ack skipped because message already acked. AckGuid={AckGuid} UserId={UserId}",
+                ackGuid,
+                userId);
+            return;
+        }
+
+        message.AckedAt = DateTimeOffset.UtcNow;
+        message.AckedBy = userId;
+        await repository.UpdateAsync(message);
+
+        logger?.LogInformation(
+            "SignalR ack accepted. AckGuid={AckGuid} UserId={UserId}",
+            ackGuid,
+            userId);
+    }
+
+    private static bool IsAckAuthorized(string messageUserId, string userId, out string reason)
+    {
+        reason = string.Empty;
+        if (string.IsNullOrWhiteSpace(messageUserId))
+        {
+            reason = "UserMissing";
+            return false;
+        }
+
+        if (!string.Equals(messageUserId, userId, StringComparison.OrdinalIgnoreCase))
+        {
+            reason = "UserMismatch";
+            return false;
+        }
+
+        return true;
+    }
 }

api/Hubs/NotificationsHub.cs

@@ -0,0 +1,6 @@
+namespace Scv.Api.Infrastructure.Options;
+
+public sealed class JobsCleanupSignalRMessagesOptions
+{
+    public string CronSchedule { get; set; } = "0 0 * * *"; // Every day at midnight
+}

api/Infrastructure/Options/CleanupSignalRMessagesJobOptions.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Linq;
+using System.Collections.Generic;
 using System.Net.Http;
 using System.Reflection;
 using Amazon;
@@ -24,6 +25,7 @@
 using Microsoft.Extensions.Options;
 using Microsoft.Graph;
 using MongoDB.Driver;
+using PostgreSQL.ListenNotify.DependencyInjection;
 using Scv.Api.Documents;
 using Scv.Api.Documents.Parsers;
 using Scv.Api.Documents.Strategies;
@@ -39,6 +41,7 @@
 using Scv.Api.Models.AccessControlManagement;
 using Scv.Api.Processors;
 using Scv.Api.Repositories;
+using Scv.Api.SignalR;
 using Scv.Api.Services;
 using Scv.Api.Services.Files;
 using Scv.Db.Contexts;
@@ -158,6 +161,46 @@ public static IServiceCollection AddGraphService(this IServiceCollection service
             return services;
         }
 
+        public static IServiceCollection AddSignalRPostgresBackplane(this IServiceCollection services, IConfiguration configuration)
+        {
+            var connectionString = configuration.GetValue<string>("DatabaseConnectionString");
+            if (string.IsNullOrWhiteSpace(connectionString))
+            {
+                throw new InvalidOperationException("DatabaseConnectionString is required to use the SignalR Postgres backplane.");
+            }
+
+            var channel = configuration.GetValue<string>("SignalR:PostgresChannel") ?? "signalr";
+            var instanceId = $"{Environment.MachineName}-{Guid.NewGuid()}";
+            var maxPayloadBytes = configuration.GetValue<int?>("SignalR:PostgresMaxPayloadBytes") ?? 7000;
+            var outboxPollSeconds = configuration.GetValue<int?>("SignalR:PostgresOutboxPollSeconds") ?? 10;
+            var outboxBatchSize = configuration.GetValue<int?>("SignalR:PostgresOutboxBatchSize") ?? 100;
+            var outboxRetentionMinutes = configuration.GetValue<int?>("SignalR:PostgresOutboxRetentionMinutes") ?? 1440;
+            var outboxMinAgeSeconds = configuration.GetValue<int?>("SignalR:PostgresOutboxMinAgeSeconds") ?? 10;
+
+            services.AddPostgresNotifications(options =>
+            {
+                options.ConnectionString = connectionString;
+                options.ListenChannels = new List<string> { channel };
+                options.DefaultNotifyChannel = channel;
+                options.ApplicationName = "Jasper.SignalR";
+            });
+
+            services.AddSingleton(new PostgresSignalRBackplaneOptions(
+                channel,
+                instanceId,
+                maxPayloadBytes,
+                outboxPollSeconds,
+                outboxBatchSize,
+                outboxRetentionMinutes,
+                outboxMinAgeSeconds));
+            services.AddSingleton<UserConnectionTracker>();
+            services.AddScoped<OrderReceivedAckNotification>();
+            services.AddScoped<INotificationService, PostgresSignalRNotificationPublisher>();
+            services.AddHostedService<PostgresSignalRNotificationBridge>();
+
+            return services;
+        }
+
         public static IServiceCollection AddAWSConfig(this IServiceCollection services, IConfiguration configuration)
         {
             var region = configuration.GetValue<string>("AWS_REGION");

api/Infrastructure/ServiceCollectionExtensions.cs

@@ -0,0 +1,48 @@
+using System;
+using System.Threading.Tasks;
+using LazyCache;
+using MapsterMapper;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Scv.Api.Infrastructure.Options;
+using Scv.Db.Repositories;
+
+namespace Scv.Api.Jobs;
+
+public class CleanupSignalRMessagesJob(
+    IConfiguration configuration,
+    IAppCache cache,
+    IMapper mapper,
+    ILogger<CleanupSignalRMessagesJob> logger,
+    INotificationRepository notificationRepository,
+    IOptions<JobsCleanupSignalRMessagesOptions> options)
+    : RecurringJobBase<CleanupSignalRMessagesJob>(configuration, cache, mapper, logger)
+{
+    private readonly INotificationRepository _notificationRepository = notificationRepository;
+    private readonly JobsCleanupSignalRMessagesOptions _options = options.Value;
+
+    public override string JobName => nameof(CleanupSignalRMessagesJob);
+
+    public override string CronSchedule => _options.CronSchedule;
+
+    public override async Task Execute()
+    {
+        var retentionMinutes = Configuration.GetValue<int>("SignalR:PostgresOutboxRetentionMinutes");
+        if (retentionMinutes <= 0)
+        {
+            Logger.LogInformation(
+                "SignalR cleanup skipped because retention minutes is {RetentionMinutes}.",
+                retentionMinutes);
+            return;
+        }
+
+        var cutoff = DateTimeOffset.UtcNow.AddMinutes(-retentionMinutes);
+        var deleted = await _notificationRepository.DeleteOlderThanAsync(cutoff);
+
+        Logger.LogInformation(
+            "SignalR cleanup deleted {DeletedCount} messages older than {Cutoff}.",
+            deleted,
+            cutoff);
+    }
+}