feat(notification-outbox): per-site KPI aggregation in the repository
This commit is contained in:
@@ -174,6 +174,68 @@ public class NotificationOutboxRepository : INotificationOutboxRepository
|
|||||||
OldestPendingAge: oldestPendingAge);
|
OldestPendingAge: oldestPendingAge);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public async Task<IReadOnlyList<SiteNotificationKpiSnapshot>> ComputePerSiteKpisAsync(
|
||||||
|
DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
var now = DateTimeOffset.UtcNow;
|
||||||
|
|
||||||
|
var queueDepth = await CountBySiteAsync(
|
||||||
|
n => n.Status == NotificationStatus.Pending || n.Status == NotificationStatus.Retrying,
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
|
var stuck = await CountBySiteAsync(
|
||||||
|
n => (n.Status == NotificationStatus.Pending || n.Status == NotificationStatus.Retrying)
|
||||||
|
&& n.CreatedAt < stuckCutoff,
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
|
var parked = await CountBySiteAsync(
|
||||||
|
n => n.Status == NotificationStatus.Parked, cancellationToken);
|
||||||
|
|
||||||
|
var delivered = await CountBySiteAsync(
|
||||||
|
n => n.Status == NotificationStatus.Delivered
|
||||||
|
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince,
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
|
// Oldest non-terminal CreatedAt per site. A SQL Min over the DateTimeOffset
|
||||||
|
// converter is awkward (see ComputeKpisAsync), so project the non-terminal
|
||||||
|
// (site, created) pairs — the live queue, which stays bounded — and reduce
|
||||||
|
// in memory.
|
||||||
|
var oldest = (await _context.Notifications
|
||||||
|
.Where(n => n.Status == NotificationStatus.Pending
|
||||||
|
|| n.Status == NotificationStatus.Retrying)
|
||||||
|
.Select(n => new { n.SourceSiteId, n.CreatedAt })
|
||||||
|
.ToListAsync(cancellationToken))
|
||||||
|
.GroupBy(x => x.SourceSiteId)
|
||||||
|
.ToDictionary(g => g.Key, g => g.Min(x => x.CreatedAt));
|
||||||
|
|
||||||
|
var siteIds = queueDepth.Keys
|
||||||
|
.Concat(stuck.Keys).Concat(parked.Keys).Concat(delivered.Keys)
|
||||||
|
.Distinct()
|
||||||
|
.OrderBy(s => s, StringComparer.Ordinal);
|
||||||
|
|
||||||
|
return siteIds.Select(site => new SiteNotificationKpiSnapshot(
|
||||||
|
SourceSiteId: site,
|
||||||
|
QueueDepth: queueDepth.GetValueOrDefault(site),
|
||||||
|
StuckCount: stuck.GetValueOrDefault(site),
|
||||||
|
ParkedCount: parked.GetValueOrDefault(site),
|
||||||
|
DeliveredLastInterval: delivered.GetValueOrDefault(site),
|
||||||
|
OldestPendingAge: oldest.TryGetValue(site, out var createdAt)
|
||||||
|
? now - createdAt
|
||||||
|
: null)).ToList();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Counts notification rows matching <paramref name="predicate"/>, grouped by source site.</summary>
|
||||||
|
private async Task<Dictionary<string, int>> CountBySiteAsync(
|
||||||
|
System.Linq.Expressions.Expression<Func<Notification, bool>> predicate,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
return await _context.Notifications
|
||||||
|
.Where(predicate)
|
||||||
|
.GroupBy(n => n.SourceSiteId)
|
||||||
|
.Select(g => new { Site = g.Key, Count = g.Count() })
|
||||||
|
.ToDictionaryAsync(x => x.Site, x => x.Count, cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
public async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
|
public async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
|
||||||
=> await _context.SaveChangesAsync(cancellationToken);
|
=> await _context.SaveChangesAsync(cancellationToken);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,70 @@
|
|||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using ScadaLink.Commons.Entities.Notifications;
|
||||||
|
using ScadaLink.Commons.Types.Enums;
|
||||||
|
using ScadaLink.ConfigurationDatabase;
|
||||||
|
using ScadaLink.ConfigurationDatabase.Repositories;
|
||||||
|
|
||||||
|
namespace ScadaLink.ConfigurationDatabase.Tests;
|
||||||
|
|
||||||
|
// Coverage for per-site KPI aggregation in the Notification Outbox repository
|
||||||
|
// (Task 2 of the notifications-nav-group feature).
|
||||||
|
public class NotificationOutboxRepositoryPerSiteKpiTests
|
||||||
|
{
|
||||||
|
private static ScadaLinkDbContext NewContext() => SqliteTestHelper.CreateInMemoryContext();
|
||||||
|
|
||||||
|
private static Notification NewNotification(
|
||||||
|
string sourceSiteId,
|
||||||
|
NotificationStatus status,
|
||||||
|
DateTimeOffset createdAt,
|
||||||
|
DateTimeOffset? deliveredAt = null)
|
||||||
|
{
|
||||||
|
return new Notification(
|
||||||
|
Guid.NewGuid().ToString(), NotificationType.Email, "Ops List", "Subject", "Body", sourceSiteId)
|
||||||
|
{
|
||||||
|
Status = status,
|
||||||
|
CreatedAt = createdAt,
|
||||||
|
DeliveredAt = deliveredAt,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ComputePerSiteKpisAsync_AggregatesMetricsPerSite()
|
||||||
|
{
|
||||||
|
await using var ctx = NewContext();
|
||||||
|
var now = DateTimeOffset.UtcNow;
|
||||||
|
|
||||||
|
// plant-a: 1 pending (stuck, created 20m ago), 1 parked
|
||||||
|
ctx.Notifications.Add(NewNotification("plant-a", NotificationStatus.Pending, createdAt: now.AddMinutes(-20)));
|
||||||
|
ctx.Notifications.Add(NewNotification("plant-a", NotificationStatus.Parked, createdAt: now.AddMinutes(-5)));
|
||||||
|
// plant-b: 1 delivered in-window, 1 pending (fresh)
|
||||||
|
ctx.Notifications.Add(NewNotification("plant-b", NotificationStatus.Delivered, createdAt: now.AddHours(-2), deliveredAt: now.AddMinutes(-2)));
|
||||||
|
ctx.Notifications.Add(NewNotification("plant-b", NotificationStatus.Pending, createdAt: now.AddMinutes(-1)));
|
||||||
|
await ctx.SaveChangesAsync();
|
||||||
|
|
||||||
|
var repo = new NotificationOutboxRepository(ctx);
|
||||||
|
var result = await repo.ComputePerSiteKpisAsync(
|
||||||
|
stuckCutoff: now.AddMinutes(-10), deliveredSince: now.AddMinutes(-30));
|
||||||
|
|
||||||
|
var a = result.Single(s => s.SourceSiteId == "plant-a");
|
||||||
|
Assert.Equal(1, a.QueueDepth);
|
||||||
|
Assert.Equal(1, a.StuckCount);
|
||||||
|
Assert.Equal(1, a.ParkedCount);
|
||||||
|
Assert.Equal(0, a.DeliveredLastInterval);
|
||||||
|
Assert.NotNull(a.OldestPendingAge);
|
||||||
|
|
||||||
|
var b = result.Single(s => s.SourceSiteId == "plant-b");
|
||||||
|
Assert.Equal(1, b.QueueDepth);
|
||||||
|
Assert.Equal(0, b.StuckCount);
|
||||||
|
Assert.Equal(1, b.DeliveredLastInterval);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ComputePerSiteKpisAsync_ReturnsEmpty_WhenNoNotifications()
|
||||||
|
{
|
||||||
|
await using var ctx = NewContext();
|
||||||
|
var repo = new NotificationOutboxRepository(ctx);
|
||||||
|
var result = await repo.ComputePerSiteKpisAsync(
|
||||||
|
DateTimeOffset.UtcNow, DateTimeOffset.UtcNow.AddMinutes(-30));
|
||||||
|
Assert.Empty(result);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user