feat(notification-outbox): async Notify.Send with status handle
Notify.To(list).Send(subject,body) now generates a NotificationId GUID, enqueues a Notification-category message into the site Store-and-Forward Engine, and returns the NotificationId immediately (Task<string>). The NotificationId is the single idempotency key end-to-end: it is the S&F message Id, it is carried inside the buffered NotificationSubmit payload, and it is the id the forwarder submits to central. NotificationForwarder now deserializes the buffered payload as a NotificationSubmit and reads NotificationId from it (re-stamping only the site-owned SourceSiteId / SourceInstanceId), instead of deriving the id from StoreAndForwardMessage.Id. Adds NotifyHelper.Status(id): queries central via the site communication actor; reports the site-local Forwarding state while the notification is still buffered at the site, maps central's response when found, and Unknown otherwise. Adds a NotificationDeliveryStatus record. SiteCommunicationActor gains a NotificationStatusQuery forwarding handler mirroring NotificationSubmit. StoreAndForwardService.EnqueueAsync gains an optional messageId parameter and exposes GetMessageByIdAsync.
This commit is contained in:
Joseph Doherty
@@ -155,6 +155,51 @@ public class SiteCommunicationActorTests : TestKit
|
||||
ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-2" && !ack.Accepted);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NotificationStatusQuery_WithCentralClient_ForwardedToCentralAndResponseRoutedBack()
|
||||
{
|
||||
// Notify.Status(id) issues a NotificationStatusQuery; the site actor forwards it
|
||||
// to central over the ClusterClient command/control transport and the central
|
||||
// response must route back to the original sender (the helper's Ask).
|
||||
var dmProbe = CreateTestProbe();
|
||||
var centralClientProbe = CreateTestProbe();
|
||||
var siteActor = Sys.ActorOf(Props.Create(() =>
|
||||
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
|
||||
|
||||
siteActor.Tell(new RegisterCentralClient(centralClientProbe.Ref));
|
||||
|
||||
var query = new NotificationStatusQuery("corr-99", "notif-1");
|
||||
siteActor.Tell(query);
|
||||
|
||||
var send = centralClientProbe.FishForMessage<ClusterClient.Send>(
|
||||
s => s.Message is NotificationStatusQuery);
|
||||
Assert.Equal("/user/central-communication", send.Path);
|
||||
var forwarded = Assert.IsType<NotificationStatusQuery>(send.Message);
|
||||
Assert.Equal("notif-1", forwarded.NotificationId);
|
||||
|
||||
// The response is sent to the ClusterClient.Send's Sender — replying as that
|
||||
// probe must land back at the test actor (the original Tell sender).
|
||||
centralClientProbe.Reply(new NotificationStatusResponse(
|
||||
"corr-99", Found: true, Status: "Delivered", RetryCount: 0,
|
||||
LastError: null, DeliveredAt: DateTimeOffset.UtcNow));
|
||||
ExpectMsg<NotificationStatusResponse>(r => r.CorrelationId == "corr-99" && r.Found);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NotificationStatusQuery_WithoutCentralClient_RepliesWithNotFound()
|
||||
{
|
||||
// No ClusterClient registered yet: the query cannot reach central, so the actor
|
||||
// replies Found: false. Notify.Status then falls back to the site S&F buffer.
|
||||
var dmProbe = CreateTestProbe();
|
||||
var siteActor = Sys.ActorOf(Props.Create(() =>
|
||||
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
|
||||
|
||||
siteActor.Tell(new NotificationStatusQuery("corr-100", "notif-2"));
|
||||
|
||||
ExpectMsg<NotificationStatusResponse>(
|
||||
r => r.CorrelationId == "corr-100" && !r.Found);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void EventLogQuery_WithoutHandler_ReturnsFailure()
|
||||
{
|
||||
|
||||
@@ -156,15 +156,26 @@ public class IntegrationSurfaceTests
|
||||
[Fact]
|
||||
public async Task ScriptContext_Notify_Send_Wired()
|
||||
{
|
||||
var mockNotify = Substitute.For<INotificationDeliveryService>();
|
||||
mockNotify.SendAsync("ops", "Alert", "Body", Arg.Any<string?>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new NotificationResult(true, null));
|
||||
// Notification Outbox: Notify.Send enqueues into the site Store-and-Forward
|
||||
// Engine and returns the NotificationId handle immediately.
|
||||
var dbName = $"NotifyWired_{Guid.NewGuid():N}";
|
||||
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
|
||||
using var keepAlive = new Microsoft.Data.Sqlite.SqliteConnection(connStr);
|
||||
keepAlive.Open();
|
||||
var storage = new StoreAndForward.StoreAndForwardStorage(
|
||||
connStr, Microsoft.Extensions.Logging.Abstractions.NullLogger<StoreAndForward.StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
var saf = new StoreAndForward.StoreAndForwardService(
|
||||
storage, new StoreAndForward.StoreAndForwardOptions(),
|
||||
Microsoft.Extensions.Logging.Abstractions.NullLogger<StoreAndForward.StoreAndForwardService>.Instance);
|
||||
|
||||
var context = CreateMinimalScriptContext(notificationService: mockNotify);
|
||||
var context = CreateMinimalScriptContext(storeAndForward: saf);
|
||||
|
||||
var result = await context.Notify.To("ops").Send("Alert", "Body");
|
||||
var notificationId = await context.Notify.To("ops").Send("Alert", "Body");
|
||||
|
||||
Assert.True(result.Success);
|
||||
Assert.False(string.IsNullOrEmpty(notificationId));
|
||||
var buffered = await saf.GetMessageByIdAsync(notificationId);
|
||||
Assert.NotNull(buffered);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -188,6 +199,7 @@ public class IntegrationSurfaceTests
|
||||
[Fact]
|
||||
public async Task ScriptContext_Notify_NoService_Throws()
|
||||
{
|
||||
// No Store-and-Forward Engine wired → Notify.Send cannot enqueue and throws.
|
||||
var context = CreateMinimalScriptContext();
|
||||
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
@@ -197,7 +209,7 @@ public class IntegrationSurfaceTests
|
||||
private static SiteRuntime.Scripts.ScriptRuntimeContext CreateMinimalScriptContext(
|
||||
IExternalSystemClient? externalSystemClient = null,
|
||||
IDatabaseGateway? databaseGateway = null,
|
||||
INotificationDeliveryService? notificationService = null)
|
||||
StoreAndForward.StoreAndForwardService? storeAndForward = null)
|
||||
{
|
||||
// Create a minimal context — we use Substitute.For<IActorRef> which is fine since
|
||||
// we won't exercise Akka functionality in these tests.
|
||||
@@ -219,6 +231,7 @@ public class IntegrationSurfaceTests
|
||||
logger: Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance,
|
||||
externalSystemClient: externalSystemClient,
|
||||
databaseGateway: databaseGateway,
|
||||
notificationService: notificationService);
|
||||
storeAndForward: storeAndForward,
|
||||
siteId: "test-site");
|
||||
}
|
||||
}
|
||||
|
||||
181
tests/ScadaLink.SiteRuntime.Tests/Scripts/NotifyHelperTests.cs
Normal file
181
tests/ScadaLink.SiteRuntime.Tests/Scripts/NotifyHelperTests.cs
Normal file
@@ -0,0 +1,181 @@
|
||||
using System.Text.Json;
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ScadaLink.Commons.Messages.Notification;
|
||||
using ScadaLink.Commons.Types.Enums;
|
||||
using ScadaLink.SiteRuntime.Scripts;
|
||||
using ScadaLink.StoreAndForward;
|
||||
|
||||
namespace ScadaLink.SiteRuntime.Tests.Scripts;
|
||||
|
||||
/// <summary>
|
||||
/// Notification Outbox (Task 19): tests for the async <c>Notify.Send</c> /
|
||||
/// <c>Notify.Status</c> script API.
|
||||
///
|
||||
/// In the outbox design <c>Notify.To("list").Send(...)</c> no longer delivers email
|
||||
/// inline — it generates a stable <c>NotificationId</c>, enqueues a
|
||||
/// <see cref="StoreAndForwardCategory.Notification"/> message into the site
|
||||
/// Store-and-Forward Engine (which Task 18 retargets to forward to central), and
|
||||
/// returns the <c>NotificationId</c> immediately. <c>Notify.Status(id)</c> queries
|
||||
/// central for delivery status, reporting the site-local <c>Forwarding</c> state
|
||||
/// while the notification is still buffered at the site.
|
||||
/// </summary>
|
||||
public class NotifyHelperTests : TestKit, IAsyncLifetime, IDisposable
|
||||
{
|
||||
private readonly SqliteConnection _keepAlive;
|
||||
private readonly StoreAndForwardStorage _storage;
|
||||
private readonly StoreAndForwardService _saf;
|
||||
|
||||
public NotifyHelperTests()
|
||||
{
|
||||
var dbName = $"NotifyTests_{Guid.NewGuid():N}";
|
||||
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
|
||||
_keepAlive = new SqliteConnection(connStr);
|
||||
_keepAlive.Open();
|
||||
|
||||
_storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
var options = new StoreAndForwardOptions
|
||||
{
|
||||
DefaultRetryInterval = TimeSpan.Zero,
|
||||
DefaultMaxRetries = 3,
|
||||
RetryTimerInterval = TimeSpan.FromMinutes(10)
|
||||
};
|
||||
_saf = new StoreAndForwardService(_storage, options, NullLogger<StoreAndForwardService>.Instance);
|
||||
}
|
||||
|
||||
public async Task InitializeAsync() => await _storage.InitializeAsync();
|
||||
|
||||
public Task DisposeAsync() => Task.CompletedTask;
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
_keepAlive.Dispose();
|
||||
}
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private ScriptRuntimeContext.NotifyHelper CreateHelper(IActorRef siteCommunicationActor)
|
||||
{
|
||||
return new ScriptRuntimeContext.NotifyHelper(
|
||||
_saf,
|
||||
siteCommunicationActor,
|
||||
"site-7",
|
||||
"Plant.Pump3",
|
||||
TimeSpan.FromSeconds(3),
|
||||
NullLogger.Instance);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Send_EnqueuesNotificationIntoStoreAndForward_AndReturnsNotificationIdImmediately()
|
||||
{
|
||||
var commProbe = CreateTestProbe();
|
||||
var notify = CreateHelper(commProbe.Ref);
|
||||
|
||||
var notificationId = await notify.To("Operators").Send("Pump alarm", "Pump 3 tripped");
|
||||
|
||||
// Send returns a non-empty NotificationId string immediately (no central round-trip).
|
||||
Assert.False(string.IsNullOrEmpty(notificationId));
|
||||
|
||||
// Exactly one Notification-category message was buffered for the S&F forwarder.
|
||||
var depth = await _saf.GetBufferDepthAsync();
|
||||
Assert.Equal(1, depth.GetValueOrDefault(StoreAndForwardCategory.Notification));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Send_BufferedPayload_CarriesListSubjectBodyAndNotificationId()
|
||||
{
|
||||
var commProbe = CreateTestProbe();
|
||||
var notify = CreateHelper(commProbe.Ref);
|
||||
|
||||
var notificationId = await notify.To("Operators").Send("Pump alarm", "Pump 3 tripped");
|
||||
|
||||
var buffered = await _saf.GetMessageByIdAsync(notificationId);
|
||||
Assert.NotNull(buffered);
|
||||
Assert.Equal(StoreAndForwardCategory.Notification, buffered!.Category);
|
||||
Assert.Equal("Operators", buffered.Target);
|
||||
Assert.Equal("Plant.Pump3", buffered.OriginInstanceName);
|
||||
|
||||
// The S&F message Id is the NotificationId — the single idempotency key.
|
||||
Assert.Equal(notificationId, buffered.Id);
|
||||
|
||||
// The payload is a NotificationSubmit carrying the same NotificationId and the
|
||||
// list / subject / body the script supplied — the shape the forwarder reads.
|
||||
var payload = JsonSerializer.Deserialize<NotificationSubmit>(buffered.PayloadJson);
|
||||
Assert.NotNull(payload);
|
||||
Assert.Equal(notificationId, payload!.NotificationId);
|
||||
Assert.Equal("Operators", payload.ListName);
|
||||
Assert.Equal("Pump alarm", payload.Subject);
|
||||
Assert.Equal("Pump 3 tripped", payload.Body);
|
||||
Assert.Equal("Plant.Pump3", payload.SourceInstanceId);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Status_WhenStillBufferedAtSite_ReportsForwarding()
|
||||
{
|
||||
var commProbe = CreateTestProbe();
|
||||
var notify = CreateHelper(commProbe.Ref);
|
||||
|
||||
// Enqueue but never let it forward — the message stays buffered at the site.
|
||||
var notificationId = await notify.To("Operators").Send("Pump alarm", "Pump 3 tripped");
|
||||
|
||||
var statusTask = notify.Status(notificationId);
|
||||
|
||||
// The status query goes to central; central has no row for an un-forwarded
|
||||
// notification, so it answers Found: false.
|
||||
var query = await commProbe.ExpectMsgAsync<NotificationStatusQuery>();
|
||||
Assert.Equal(notificationId, query.NotificationId);
|
||||
commProbe.Reply(new NotificationStatusResponse(
|
||||
query.CorrelationId, Found: false, Status: "Unknown",
|
||||
RetryCount: 0, LastError: null, DeliveredAt: null));
|
||||
|
||||
var status = await statusTask;
|
||||
|
||||
// Found: false AND still in the site S&F buffer → the site-local Forwarding state.
|
||||
Assert.Equal("Forwarding", status.Status);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Status_WhenCentralReportsDelivered_MapsTheCentralResponse()
|
||||
{
|
||||
var commProbe = CreateTestProbe();
|
||||
var notify = CreateHelper(commProbe.Ref);
|
||||
|
||||
var deliveredAt = DateTimeOffset.UtcNow;
|
||||
var statusTask = notify.Status("not-buffered-id");
|
||||
|
||||
var query = await commProbe.ExpectMsgAsync<NotificationStatusQuery>();
|
||||
commProbe.Reply(new NotificationStatusResponse(
|
||||
query.CorrelationId, Found: true, Status: "Delivered",
|
||||
RetryCount: 2, LastError: "earlier transient", DeliveredAt: deliveredAt));
|
||||
|
||||
var status = await statusTask;
|
||||
|
||||
Assert.Equal("Delivered", status.Status);
|
||||
Assert.Equal(2, status.RetryCount);
|
||||
Assert.Equal("earlier transient", status.LastError);
|
||||
Assert.Equal(deliveredAt, status.DeliveredAt);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Status_WhenCentralNotFoundAndNotBuffered_ReportsUnknown()
|
||||
{
|
||||
var commProbe = CreateTestProbe();
|
||||
var notify = CreateHelper(commProbe.Ref);
|
||||
|
||||
var statusTask = notify.Status("never-existed-id");
|
||||
|
||||
var query = await commProbe.ExpectMsgAsync<NotificationStatusQuery>();
|
||||
commProbe.Reply(new NotificationStatusResponse(
|
||||
query.CorrelationId, Found: false, Status: "Unknown",
|
||||
RetryCount: 0, LastError: null, DeliveredAt: null));
|
||||
|
||||
var status = await statusTask;
|
||||
|
||||
// Not at central, not in the site buffer → genuinely unknown, NOT Forwarding.
|
||||
Assert.Equal("Unknown", status.Status);
|
||||
}
|
||||
}
|
||||
@@ -18,19 +18,26 @@ public class NotificationForwarderTests : TestKit
|
||||
|
||||
/// <summary>
|
||||
/// Builds a buffered notification S&F message whose payload matches the shape
|
||||
/// produced by the site NotificationDeliveryService enqueue path.
|
||||
/// produced by the site <c>Notify.Send</c> enqueue path (Task 19): a serialized
|
||||
/// <see cref="NotificationSubmit"/> carrying a script-generated
|
||||
/// <see cref="NotificationSubmit.NotificationId"/>. The S&F message
|
||||
/// <see cref="StoreAndForwardMessage.Id"/> equals that same id.
|
||||
/// </summary>
|
||||
private static StoreAndForwardMessage BufferedNotification(
|
||||
string id = "msg-1", string listName = "Operators",
|
||||
string subject = "Pump alarm", string message = "Pump 3 tripped",
|
||||
string? originInstance = "Plant.Pump3")
|
||||
string? originInstance = "Plant.Pump3", string? sourceScript = "alarmScript")
|
||||
{
|
||||
var payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
ListName = listName,
|
||||
Subject = subject,
|
||||
Message = message
|
||||
});
|
||||
var payload = JsonSerializer.Serialize(new NotificationSubmit(
|
||||
NotificationId: id,
|
||||
ListName: listName,
|
||||
Subject: subject,
|
||||
Body: message,
|
||||
// SourceSiteId is re-stamped by the forwarder; the enqueue side leaves it blank.
|
||||
SourceSiteId: string.Empty,
|
||||
SourceInstanceId: originInstance,
|
||||
SourceScript: sourceScript,
|
||||
SiteEnqueuedAt: DateTimeOffset.UtcNow));
|
||||
return new StoreAndForwardMessage
|
||||
{
|
||||
Id = id,
|
||||
@@ -57,11 +64,15 @@ public class NotificationForwarderTests : TestKit
|
||||
// The central target receives a NotificationSubmit whose fields map from the
|
||||
// buffered payload; reply Accepted so the handler completes as delivered.
|
||||
var submit = centralProbe.ExpectMsg<NotificationSubmit>();
|
||||
Assert.Equal("msg-1", submit.NotificationId);
|
||||
Assert.Equal("Operators", submit.ListName);
|
||||
Assert.Equal("Pump alarm", submit.Subject);
|
||||
Assert.Equal("Pump 3 tripped", submit.Body);
|
||||
// SourceSiteId is re-stamped by the forwarder from its own site id.
|
||||
Assert.Equal("site-7", submit.SourceSiteId);
|
||||
Assert.Equal("Plant.Pump3", submit.SourceInstanceId);
|
||||
// The originating script travels through from the buffered payload.
|
||||
Assert.Equal("alarmScript", submit.SourceScript);
|
||||
centralProbe.Reply(new NotificationSubmitAck(submit.NotificationId, Accepted: true, Error: null));
|
||||
|
||||
Assert.True(await deliverTask);
|
||||
@@ -76,12 +87,15 @@ public class NotificationForwarderTests : TestKit
|
||||
|
||||
// A buffered payload carrying an empty-string ListName: the empty value must not
|
||||
// be forwarded — the forwarder falls back to the S&F message Target instead.
|
||||
var payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
ListName = "",
|
||||
Subject = "Pump alarm",
|
||||
Message = "Pump 3 tripped"
|
||||
});
|
||||
var payload = JsonSerializer.Serialize(new NotificationSubmit(
|
||||
NotificationId: "msg-empty-list",
|
||||
ListName: "",
|
||||
Subject: "Pump alarm",
|
||||
Body: "Pump 3 tripped",
|
||||
SourceSiteId: string.Empty,
|
||||
SourceInstanceId: "Plant.Pump3",
|
||||
SourceScript: null,
|
||||
SiteEnqueuedAt: DateTimeOffset.UtcNow));
|
||||
var msg = new StoreAndForwardMessage
|
||||
{
|
||||
Id = "msg-empty-list",
|
||||
|
||||
Reference in New Issue
Block a user