SingalR for realtime entitychanges
All checks were successful
Build Docker Images / test (map[name:Morska plugin_project:DiunaBI.Plugins.Morska]) (push) Successful in 1m36s
Build Docker Images / test (map[name:PedrolloPL plugin_project:DiunaBI.Plugins.PedrolloPL]) (push) Successful in 1m31s
Build Docker Images / build-and-push (map[image_suffix:morska name:Morska plugin_project:DiunaBI.Plugins.Morska]) (push) Successful in 1m55s
Build Docker Images / build-and-push (map[image_suffix:pedrollopl name:PedrolloPL plugin_project:DiunaBI.Plugins.PedrolloPL]) (push) Successful in 1m53s

This commit is contained in:
2025-12-04 22:20:00 +01:00
parent bf2beda390
commit 193127b86a
11 changed files with 509 additions and 6 deletions

View File

@@ -0,0 +1,16 @@
using Microsoft.AspNetCore.SignalR;
namespace DiunaBI.API.Hubs;
public class EntityChangeHub : Hub
{
public async Task SendEntityChange(string module, string id, string operation)
{
await Clients.All.SendAsync("EntityChanged", new
{
module,
id,
operation
});
}
}

View File

@@ -4,8 +4,10 @@ using Microsoft.IdentityModel.Tokens;
using System.IdentityModel.Tokens.Jwt; using System.IdentityModel.Tokens.Jwt;
using System.Reflection; using System.Reflection;
using System.Text; using System.Text;
using DiunaBI.API.Hubs;
using DiunaBI.API.Services; using DiunaBI.API.Services;
using DiunaBI.Infrastructure.Data; using DiunaBI.Infrastructure.Data;
using DiunaBI.Infrastructure.Interceptors;
using DiunaBI.Infrastructure.Services; using DiunaBI.Infrastructure.Services;
using Google.Apis.Sheets.v4; using Google.Apis.Sheets.v4;
using Serilog; using Serilog;
@@ -29,10 +31,17 @@ if (builder.Environment.IsProduction())
var connectionString = builder.Configuration.GetConnectionString("SQLDatabase"); var connectionString = builder.Configuration.GetConnectionString("SQLDatabase");
builder.Services.AddDbContext<AppDbContext>(x => // Register EntityChangeInterceptor
builder.Services.AddSingleton<EntityChangeInterceptor>();
builder.Services.AddDbContext<AppDbContext>((serviceProvider, options) =>
{ {
x.UseSqlServer(connectionString, sqlOptions => sqlOptions.MigrationsAssembly("DiunaBI.Infrastructure")); options.UseSqlServer(connectionString, sqlOptions => sqlOptions.MigrationsAssembly("DiunaBI.Infrastructure"));
x.EnableSensitiveDataLogging(); options.EnableSensitiveDataLogging();
// Add EntityChangeInterceptor
var interceptor = serviceProvider.GetRequiredService<EntityChangeInterceptor>();
options.AddInterceptors(interceptor);
}); });
builder.Services.AddCors(options => builder.Services.AddCors(options =>
@@ -58,6 +67,9 @@ builder.Services.AddCors(options =>
builder.Services.AddControllers(); builder.Services.AddControllers();
// SignalR
builder.Services.AddSignalR();
builder.Services.AddAuthentication(options => builder.Services.AddAuthentication(options =>
{ {
options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme; options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
@@ -244,6 +256,9 @@ app.Use(async (context, next) =>
app.MapControllers(); app.MapControllers();
// SignalR Hub
app.MapHub<EntityChangeHub>("/hubs/entitychanges");
app.MapGet("/health", () => Results.Ok(new { status = "OK", timestamp = DateTime.UtcNow })) app.MapGet("/health", () => Results.Ok(new { status = "OK", timestamp = DateTime.UtcNow }))
.AllowAnonymous(); .AllowAnonymous();

View File

@@ -26,4 +26,8 @@
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" /> <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
</ItemGroup> </ItemGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup>
</Project> </Project>

View File

@@ -0,0 +1,201 @@
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
namespace DiunaBI.Infrastructure.Interceptors;
public class EntityChangeInterceptor : SaveChangesInterceptor
{
private readonly object? _hubContext;
private readonly ILogger<EntityChangeInterceptor>? _logger;
private readonly List<(string Module, string Id, string Operation)> _pendingChanges = new();
public EntityChangeInterceptor(IServiceProvider serviceProvider)
{
_logger = serviceProvider.GetService(typeof(ILogger<EntityChangeInterceptor>)) as ILogger<EntityChangeInterceptor>;
// Try to get hub context - it may not be registered in some scenarios (e.g., migrations)
try
{
var hubType = Type.GetType("DiunaBI.API.Hubs.EntityChangeHub, DiunaBI.API");
if (hubType != null)
{
var hubContextType = typeof(IHubContext<>).MakeGenericType(hubType);
_hubContext = serviceProvider.GetService(hubContextType);
if (_hubContext != null)
{
_logger?.LogInformation("✅ EntityChangeInterceptor: Hub context initialized");
Console.WriteLine("✅ EntityChangeInterceptor: Hub context initialized");
}
else
{
_logger?.LogWarning("⚠️ EntityChangeInterceptor: Hub context is null");
Console.WriteLine("⚠️ EntityChangeInterceptor: Hub context is null");
}
}
else
{
_logger?.LogWarning("⚠️ EntityChangeInterceptor: Hub type not found");
Console.WriteLine("⚠️ EntityChangeInterceptor: Hub type not found");
}
}
catch (Exception ex)
{
_logger?.LogError(ex, "❌ EntityChangeInterceptor: Failed to initialize hub context");
Console.WriteLine($"❌ EntityChangeInterceptor: Failed to initialize hub context: {ex.Message}");
_hubContext = null;
}
}
public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
DbContextEventData eventData,
InterceptionResult<int> result,
CancellationToken cancellationToken = default)
{
_pendingChanges.Clear();
Console.WriteLine($"🔍 EntityChangeInterceptor.SavingChangesAsync called. HubContext null? {_hubContext == null}, Context null? {eventData.Context == null}");
if (_hubContext != null && eventData.Context != null)
{
// Capture changes BEFORE save
var entries = eventData.Context.ChangeTracker.Entries().ToList();
Console.WriteLine($"🔍 Found {entries.Count} total entries in ChangeTracker");
foreach (var entry in entries)
{
Console.WriteLine($"🔍 Entry: {entry.Metadata.ClrType.Name}, State: {entry.State}");
if (entry.State == EntityState.Added ||
entry.State == EntityState.Modified ||
entry.State == EntityState.Deleted)
{
var module = entry.Metadata.GetTableName() ?? entry.Metadata.ClrType.Name;
var id = GetEntityId(entry);
var operation = entry.State switch
{
EntityState.Added => "created",
EntityState.Modified => "updated",
EntityState.Deleted => "deleted",
_ => "unknown"
};
Console.WriteLine($"🔍 Detected change: {module} {id} {operation}");
if (id != null)
{
_pendingChanges.Add((module, id, operation));
Console.WriteLine($"✅ Added to pending changes: {module} {id} {operation}");
}
else
{
Console.WriteLine($"⚠️ Skipped (id is null): {module} {operation}");
}
}
}
Console.WriteLine($"🔍 Total pending changes: {_pendingChanges.Count}");
}
return base.SavingChangesAsync(eventData, result, cancellationToken);
}
public override async ValueTask<int> SavedChangesAsync(
SaveChangesCompletedEventData eventData,
int result,
CancellationToken cancellationToken = default)
{
// Broadcast changes AFTER successful save
if (_hubContext != null && result > 0 && _pendingChanges.Any())
{
_logger?.LogInformation("📤 Broadcasting {Count} entity changes via SignalR", _pendingChanges.Count);
Console.WriteLine($"📤 Broadcasting {_pendingChanges.Count} entity changes via SignalR");
foreach (var (module, id, operation) in _pendingChanges)
{
try
{
Console.WriteLine($"📤 Broadcasting: {module} {id} {operation}");
// Use reflection to call hub methods since we can't reference the API project
var clientsProperty = _hubContext.GetType().GetProperty("Clients");
Console.WriteLine($" 🔍 Clients property: {clientsProperty != null}");
if (clientsProperty != null)
{
var clients = clientsProperty.GetValue(_hubContext);
Console.WriteLine($" 🔍 Clients value: {clients != null}, Type: {clients?.GetType().Name}");
if (clients != null)
{
var allProperty = clients.GetType().GetProperty("All");
Console.WriteLine($" 🔍 All property: {allProperty != null}");
if (allProperty != null)
{
var allClients = allProperty.GetValue(clients);
Console.WriteLine($" 🔍 AllClients value: {allClients != null}, Type: {allClients?.GetType().Name}");
if (allClients != null)
{
// SendAsync is an extension method, so we need to find it differently
// Look for the IClientProxy interface which has SendCoreAsync
var sendCoreAsyncMethod = allClients.GetType().GetMethod("SendCoreAsync");
Console.WriteLine($" 🔍 SendCoreAsync method found: {sendCoreAsyncMethod != null}");
if (sendCoreAsyncMethod != null)
{
// SendCoreAsync takes (string method, object?[] args, CancellationToken cancellationToken)
var task = sendCoreAsyncMethod.Invoke(allClients, new object[]
{
"EntityChanged",
new object[] { new { module, id, operation } },
cancellationToken
}) as Task;
Console.WriteLine($" 🔍 Task created: {task != null}");
if (task != null)
{
await task;
Console.WriteLine($"✅ Broadcast successful: {module} {id} {operation}");
}
else
{
Console.WriteLine($"❌ Task is null after invoke");
}
}
else
{
Console.WriteLine($"❌ SendCoreAsync method not found");
}
}
}
}
}
}
catch (Exception ex)
{
_logger?.LogError(ex, "❌ Failed to broadcast entity change");
Console.WriteLine($"❌ Failed to broadcast: {ex.Message}");
Console.WriteLine($"❌ Stack trace: {ex.StackTrace}");
}
}
}
_pendingChanges.Clear();
return await base.SavedChangesAsync(eventData, result, cancellationToken);
}
private static string? GetEntityId(Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry entry)
{
var keyProperty = entry.Metadata.FindPrimaryKey()?.Properties.FirstOrDefault();
if (keyProperty == null)
return null;
var value = entry.Property(keyProperty.Name).CurrentValue;
return value?.ToString();
}
}

View File

@@ -8,9 +8,10 @@ using Microsoft.JSInterop;
namespace DiunaBI.UI.Shared.Components; namespace DiunaBI.UI.Shared.Components;
public partial class JobListComponent : ComponentBase public partial class JobListComponent : ComponentBase, IDisposable
{ {
[Inject] private JobService JobService { get; set; } = default!; [Inject] private JobService JobService { get; set; } = default!;
[Inject] private EntityChangeHubService HubService { get; set; } = default!;
[Inject] private ISnackbar Snackbar { get; set; } = default!; [Inject] private ISnackbar Snackbar { get; set; } = default!;
[Inject] private NavigationManager NavigationManager { get; set; } = default!; [Inject] private NavigationManager NavigationManager { get; set; } = default!;
[Inject] private IJSRuntime JSRuntime { get; set; } = default!; [Inject] private IJSRuntime JSRuntime { get; set; } = default!;
@@ -25,6 +26,32 @@ public partial class JobListComponent : ComponentBase
protected override async Task OnInitializedAsync() protected override async Task OnInitializedAsync()
{ {
await LoadJobs(); await LoadJobs();
// Subscribe to SignalR entity changes
HubService.EntityChanged += OnEntityChanged;
}
private async void OnEntityChanged(string module, string id, string operation)
{
Console.WriteLine($"🔔 JobListComponent.OnEntityChanged called: module={module}, id={id}, operation={operation}");
// Only react if it's a QueueJobs change
if (module.Equals("QueueJobs", StringComparison.OrdinalIgnoreCase))
{
Console.WriteLine($"📨 Job {id} changed, refreshing job list");
await InvokeAsync(async () =>
{
Console.WriteLine($"🔄 LoadJobs starting...");
await LoadJobs();
Console.WriteLine($"🔄 StateHasChanged calling...");
StateHasChanged();
Console.WriteLine($"✅ Job list refresh complete");
});
}
else
{
Console.WriteLine($"⏭️ Skipping - module '{module}' is not QueueJobs");
}
} }
private async Task LoadJobs() private async Task LoadJobs()
@@ -107,4 +134,9 @@ public partial class JobListComponent : ComponentBase
_ => Color.Default _ => Color.Default
}; };
} }
public void Dispose()
{
HubService.EntityChanged -= OnEntityChanged;
}
} }

View File

@@ -17,6 +17,7 @@
<PackageReference Include="Microsoft.AspNetCore.WebUtilities" Version="10.0.0" /> <PackageReference Include="Microsoft.AspNetCore.WebUtilities" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Http" Version="10.0.0" /> <PackageReference Include="Microsoft.Extensions.Http" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.0" /> <PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="10.0.0" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>

View File

@@ -1,4 +1,6 @@
using Microsoft.AspNetCore.Components;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using DiunaBI.UI.Shared.Services; using DiunaBI.UI.Shared.Services;
using DiunaBI.UI.Shared.Handlers; using DiunaBI.UI.Shared.Handlers;
@@ -44,6 +46,15 @@ public static class ServiceCollectionExtensions
services.AddScoped<LayerFilterStateService>(); services.AddScoped<LayerFilterStateService>();
services.AddScoped<DataInboxFilterStateService>(); services.AddScoped<DataInboxFilterStateService>();
// SignalR Hub Service (singleton for global connection shared across all users)
services.AddSingleton<EntityChangeHubService>(sp =>
{
// For singleton, we can't inject scoped services directly
// We'll get them from the service provider when needed
var logger = sp.GetRequiredService<ILogger<EntityChangeHubService>>();
return new EntityChangeHubService(apiBaseUrl, sp, logger);
});
return services; return services;
} }
} }

View File

@@ -1,6 +1,7 @@
@using MudBlazor @using MudBlazor
@using DiunaBI.UI.Shared.Services @using DiunaBI.UI.Shared.Services
@inject AppConfig AppConfig @inject AppConfig AppConfig
@inject EntityChangeHubService HubService
@inherits LayoutComponentBase @inherits LayoutComponentBase
<AuthGuard> <AuthGuard>
@@ -54,6 +55,12 @@
private bool _drawerOpen = true; private bool _drawerOpen = true;
private DrawerVariant _drawerVariant = DrawerVariant.Persistent; private DrawerVariant _drawerVariant = DrawerVariant.Persistent;
protected override async Task OnInitializedAsync()
{
// Initialize SignalR connection when layout loads
await HubService.InitializeAsync();
}
private MudTheme _theme = new MudTheme() private MudTheme _theme = new MudTheme()
{ {
PaletteLight = new PaletteLight() PaletteLight = new PaletteLight()

View File

@@ -3,8 +3,10 @@
@using DiunaBI.Domain.Entities @using DiunaBI.Domain.Entities
@using MudBlazor @using MudBlazor
@inject JobService JobService @inject JobService JobService
@inject EntityChangeHubService HubService
@inject NavigationManager NavigationManager @inject NavigationManager NavigationManager
@inject ISnackbar Snackbar @inject ISnackbar Snackbar
@implements IDisposable
<MudCard> <MudCard>
<MudCardHeader> <MudCardHeader>
@@ -160,6 +162,24 @@
protected override async Task OnInitializedAsync() protected override async Task OnInitializedAsync()
{ {
await LoadJob(); await LoadJob();
// Subscribe to SignalR entity changes
HubService.EntityChanged += OnEntityChanged;
}
private async void OnEntityChanged(string module, string id, string operation)
{
// Only react if it's a QueueJobs change for this specific job
if (module.Equals("QueueJobs", StringComparison.OrdinalIgnoreCase) &&
Guid.TryParse(id, out var jobId) && jobId == Id)
{
Console.WriteLine($"📨 Job {jobId} changed, refreshing detail page");
await InvokeAsync(async () =>
{
await LoadJob();
StateHasChanged();
});
}
} }
private async Task LoadJob() private async Task LoadJob()
@@ -242,4 +262,9 @@
_ => Icons.Material.Filled.Help _ => Icons.Material.Filled.Help
}; };
} }
public void Dispose()
{
HubService.EntityChanged -= OnEntityChanged;
}
} }

View File

@@ -439,10 +439,10 @@ public partial class LayerDetailPage : ComponentBase
} }
else else
{ {
Snackbar.Add("Job created successfully!", Severity.Success); Snackbar.Add("Job created successfully! Watch real-time status updates.", Severity.Success);
} }
// Navigate to job detail page // Navigate to job detail page to see real-time updates
NavigationManager.NavigateTo($"/jobs/{result.JobId}"); NavigationManager.NavigateTo($"/jobs/{result.JobId}");
} }
else else

View File

@@ -0,0 +1,191 @@
using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.Extensions.Logging;
namespace DiunaBI.UI.Shared.Services;
public class EntityChangeHubService : IAsyncDisposable
{
private readonly string _hubUrl;
private readonly ILogger<EntityChangeHubService> _logger;
private HubConnection? _hubConnection;
private bool _isInitialized;
private readonly SemaphoreSlim _initializationLock = new SemaphoreSlim(1, 1);
private static int _instanceCounter = 0;
private readonly int _instanceId;
// Events that components can subscribe to
public event Action<string, string, string>? EntityChanged;
public EntityChangeHubService(
string apiBaseUrl,
IServiceProvider serviceProvider,
ILogger<EntityChangeHubService> logger)
{
_instanceId = Interlocked.Increment(ref _instanceCounter);
// Convert HTTP URL to SignalR hub URL
var baseUrl = apiBaseUrl.TrimEnd('/');
_hubUrl = baseUrl + "/hubs/entitychanges";
_logger = logger;
_logger.LogInformation("🏗️ EntityChangeHubService instance #{InstanceId} created. Hub URL: {HubUrl}", _instanceId, _hubUrl);
Console.WriteLine($"🏗️ EntityChangeHubService instance #{_instanceId} created. Hub URL: {_hubUrl}, _isInitialized = {_isInitialized}");
}
public async Task InitializeAsync()
{
_logger.LogInformation("🔍 Instance #{InstanceId} InitializeAsync called. _isInitialized = {IsInitialized}, _hubConnection null? {IsNull}", _instanceId, _isInitialized, _hubConnection == null);
Console.WriteLine($"🔍 Instance #{_instanceId} InitializeAsync called. _isInitialized = {_isInitialized}, _hubConnection null? {_hubConnection == null}");
if (_isInitialized)
{
_logger.LogInformation("⏭️ Instance #{InstanceId} SignalR already initialized, skipping", _instanceId);
Console.WriteLine($"⏭️ Instance #{_instanceId} SignalR already initialized, skipping");
return;
}
await _initializationLock.WaitAsync();
try
{
// Double-check after acquiring lock
if (_isInitialized)
{
Console.WriteLine($"⏭️ SignalR already initialized (after lock), skipping");
return;
}
_logger.LogInformation("🔌 Initializing SignalR connection to {HubUrl}", _hubUrl);
Console.WriteLine($"🔌 Initializing SignalR connection to {_hubUrl}");
_hubConnection = new HubConnectionBuilder()
.WithUrl(_hubUrl)
.WithAutomaticReconnect()
.Build();
// Subscribe to EntityChanged messages
_hubConnection.On<object>("EntityChanged", (data) =>
{
Console.WriteLine($"🔔 RAW SignalR message received at {DateTime.Now:HH:mm:ss.fff}");
Console.WriteLine($"🔔 Data type: {data?.GetType().FullName}");
try
{
// Parse the anonymous object
var json = System.Text.Json.JsonSerializer.Serialize(data);
Console.WriteLine($"📨 Received SignalR message: {json}");
// Use case-insensitive deserialization (backend sends camelCase: module, id, operation)
var options = new System.Text.Json.JsonSerializerOptions
{
PropertyNameCaseInsensitive = true
};
var change = System.Text.Json.JsonSerializer.Deserialize<EntityChangeMessage>(json, options);
if (change != null)
{
_logger.LogInformation("📨 Received entity change: {Module} {Id} {Operation}",
change.Module, change.Id, change.Operation);
Console.WriteLine($"📨 Entity change: {change.Module} {change.Id} {change.Operation}");
// Notify all subscribers
Console.WriteLine($"🔔 Invoking EntityChanged event, subscribers: {EntityChanged?.GetInvocationList().Length ?? 0}");
EntityChanged?.Invoke(change.Module, change.Id, change.Operation);
Console.WriteLine($"🔔 EntityChanged event invoked successfully");
}
else
{
Console.WriteLine($"⚠️ Deserialized change is null");
}
}
catch (Exception ex)
{
_logger.LogError(ex, "❌ Error processing entity change message");
Console.WriteLine($"❌ Error processing message: {ex.Message}");
Console.WriteLine($"❌ Stack trace: {ex.StackTrace}");
}
});
_hubConnection.Reconnecting += (error) =>
{
_logger.LogWarning("SignalR reconnecting: {Error}", error?.Message);
Console.WriteLine($"⚠️ SignalR reconnecting: {error?.Message}");
return Task.CompletedTask;
};
_hubConnection.Reconnected += (connectionId) =>
{
_logger.LogInformation("✅ SignalR reconnected: {ConnectionId}", connectionId);
Console.WriteLine($"✅ SignalR reconnected: {connectionId}");
return Task.CompletedTask;
};
_hubConnection.Closed += (error) =>
{
_logger.LogError(error, "❌ SignalR connection closed");
Console.WriteLine($"❌ SignalR connection closed: {error?.Message}");
return Task.CompletedTask;
};
await StartConnectionAsync();
_isInitialized = true;
_logger.LogInformation("✅ Instance #{InstanceId} _isInitialized set to true", _instanceId);
Console.WriteLine($"✅ Instance #{_instanceId} _isInitialized set to true");
}
catch (Exception ex)
{
_logger.LogError(ex, "❌ Instance #{InstanceId} Failed to initialize SignalR connection", _instanceId);
Console.WriteLine($"❌ Instance #{_instanceId} Failed to initialize SignalR: {ex.Message}");
}
finally
{
_initializationLock.Release();
}
}
private async Task StartConnectionAsync()
{
if (_hubConnection == null)
{
_logger.LogWarning("Hub connection is null, cannot start");
return;
}
try
{
Console.WriteLine($"🔌 Starting SignalR connection...");
await _hubConnection.StartAsync();
_logger.LogInformation("✅ SignalR connected successfully");
Console.WriteLine($"✅ SignalR connected successfully to {_hubUrl}");
}
catch (Exception ex)
{
_logger.LogError(ex, "❌ Failed to start SignalR connection");
Console.WriteLine($"❌ Failed to start SignalR: {ex.Message}\n{ex.StackTrace}");
}
}
public async ValueTask DisposeAsync()
{
if (_hubConnection != null)
{
try
{
await _hubConnection.StopAsync();
await _hubConnection.DisposeAsync();
}
catch (Exception ex)
{
_logger.LogError(ex, "Error disposing SignalR connection");
}
}
_initializationLock?.Dispose();
}
}
public class EntityChangeMessage
{
public string Module { get; set; } = string.Empty;
public string Id { get; set; } = string.Empty;
public string Operation { get; set; } = string.Empty;
}