feat(commons): project outbound API/DB/notification rows onto the External lane
ResolveKind maps ApiOutbound/DbOutbound/Notification to ApiCall/DbCall/Notify;
ResolveLane puts the three on FlowLane.External; outbound outcomes map
Delivered→Ok, Failed→Failed, Parked/Discarded→Parked, else Pending; subtitles
'{status} · sync|cached', '{op} · n rows' from a DbWrite Extra, '{status}' for
Notify. New FlowStepProjector.IsOutbound helper.
Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
@@ -4,12 +4,12 @@
|
|||||||
{
|
{
|
||||||
"id": 0,
|
"id": 0,
|
||||||
"subject": "Task 0: Commons \u2014 kinds, lane, outcomes, lifecycle row",
|
"subject": "Task 0: Commons \u2014 kinds, lane, outcomes, lifecycle row",
|
||||||
"status": "pending"
|
"status": "completed"
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"id": 1,
|
"id": 1,
|
||||||
"subject": "Task 1: Commons \u2014 FlowStepProjector maps outbound rows",
|
"subject": "Task 1: Commons \u2014 FlowStepProjector maps outbound rows",
|
||||||
"status": "pending",
|
"status": "in_progress",
|
||||||
"blockedBy": [
|
"blockedBy": [
|
||||||
0
|
0
|
||||||
]
|
]
|
||||||
@@ -96,5 +96,5 @@
|
|||||||
]
|
]
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
"lastUpdated": "2026-09-06T05:32:15.141670+00:00"
|
"lastUpdated": "2026-09-06T05:37:29.919766+00:00"
|
||||||
}
|
}
|
||||||
@@ -1,3 +1,4 @@
|
|||||||
|
using System.Text.Json;
|
||||||
using ZB.MOM.WW.Audit;
|
using ZB.MOM.WW.Audit;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||||
@@ -73,17 +74,34 @@ public static class FlowStepProjector
|
|||||||
return FlowStepKind.Unknown;
|
return FlowStepKind.Unknown;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>Lane rule: route call / inbound / response → Central; script run / wait → Site; write → Tags.</summary>
|
/// <summary>Lane rule: route call / inbound / response → Central; script run / wait → Site; write → Tags; outbound API / DB / notify → External.</summary>
|
||||||
public static FlowLane ResolveLane(FlowStepKind kind, string? sourceSiteId) => kind switch
|
public static FlowLane ResolveLane(FlowStepKind kind, string? sourceSiteId) => kind switch
|
||||||
{
|
{
|
||||||
FlowStepKind.InboundRequest or FlowStepKind.RouteCall or FlowStepKind.Response => FlowLane.Central,
|
FlowStepKind.InboundRequest or FlowStepKind.RouteCall or FlowStepKind.Response => FlowLane.Central,
|
||||||
FlowStepKind.ScriptRun or FlowStepKind.AttributeWait => FlowLane.Site,
|
FlowStepKind.ScriptRun or FlowStepKind.AttributeWait => FlowLane.Site,
|
||||||
FlowStepKind.AttributeWrite => FlowLane.Tags,
|
FlowStepKind.AttributeWrite => FlowLane.Tags,
|
||||||
|
FlowStepKind.ApiCall or FlowStepKind.DbCall or FlowStepKind.Notify => FlowLane.External,
|
||||||
_ => string.IsNullOrEmpty(sourceSiteId) ? FlowLane.Central : FlowLane.Site,
|
_ => string.IsNullOrEmpty(sourceSiteId) ? FlowLane.Central : FlowLane.Site,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/// <summary>True for the three outbound kinds that live on the External lane.</summary>
|
||||||
|
public static bool IsOutbound(FlowStepKind kind)
|
||||||
|
=> kind is FlowStepKind.ApiCall or FlowStepKind.DbCall or FlowStepKind.Notify;
|
||||||
|
|
||||||
private static FlowStepOutcome ResolveOutcome(FlowStepKind kind, string? status, string? error)
|
private static FlowStepOutcome ResolveOutcome(FlowStepKind kind, string? status, string? error)
|
||||||
{
|
{
|
||||||
|
if (IsOutbound(kind))
|
||||||
|
{
|
||||||
|
return status switch
|
||||||
|
{
|
||||||
|
nameof(AuditStatus.Delivered) => FlowStepOutcome.Ok,
|
||||||
|
nameof(AuditStatus.Failed) => FlowStepOutcome.Failed,
|
||||||
|
nameof(AuditStatus.Parked) or nameof(AuditStatus.Discarded) => FlowStepOutcome.Parked,
|
||||||
|
// Submitted / Forwarded / Attempted — and anything this build does not know — is still in flight.
|
||||||
|
_ => FlowStepOutcome.Pending,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
if (!string.Equals(status, nameof(AuditStatus.Failed), StringComparison.Ordinal))
|
if (!string.Equals(status, nameof(AuditStatus.Failed), StringComparison.Ordinal))
|
||||||
return FlowStepOutcome.Ok;
|
return FlowStepOutcome.Ok;
|
||||||
if ((kind == FlowStepKind.ScriptRun || kind == FlowStepKind.AttributeWait)
|
if ((kind == FlowStepKind.ScriptRun || kind == FlowStepKind.AttributeWait)
|
||||||
@@ -107,6 +125,45 @@ public static class FlowStepProjector
|
|||||||
: extra.WaitOutcome is null
|
: extra.WaitOutcome is null
|
||||||
? extra.ObservedValue
|
? extra.ObservedValue
|
||||||
: extra.ObservedValue is null ? extra.WaitOutcome : $"{extra.WaitOutcome} · {extra.ObservedValue}",
|
: extra.ObservedValue is null ? extra.WaitOutcome : $"{extra.WaitOutcome} · {extra.ObservedValue}",
|
||||||
|
FlowStepKind.ApiCall => string.Equals(d.Kind, nameof(AuditKind.ApiCall), StringComparison.Ordinal)
|
||||||
|
? $"{d.Status} · sync"
|
||||||
|
: $"{d.Status} · cached",
|
||||||
|
FlowStepKind.DbCall => DbSubtitle(d),
|
||||||
|
FlowStepKind.Notify => d.Status,
|
||||||
_ => d.Kind,
|
_ => d.Kind,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A sync <c>DbWrite</c> row carries <c>{"op":"write","rowsAffected":n}</c> / <c>{"op":"read","rowsReturned":n}</c>
|
||||||
|
/// in <c>Extra</c> (site <c>AuditingDbCommand</c> vocabulary); cached rows carry none and read as <c>"{status} · cached"</c>.
|
||||||
|
/// </summary>
|
||||||
|
private static string? DbSubtitle(AuditDetails d)
|
||||||
|
{
|
||||||
|
if (string.Equals(d.Kind, nameof(AuditKind.DbWrite), StringComparison.Ordinal) && !string.IsNullOrWhiteSpace(d.Extra))
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
using var doc = JsonDocument.Parse(d.Extra);
|
||||||
|
var root = doc.RootElement;
|
||||||
|
if (root.ValueKind == JsonValueKind.Object && root.TryGetProperty("op", out var opEl) && opEl.ValueKind == JsonValueKind.String)
|
||||||
|
{
|
||||||
|
var op = opEl.GetString();
|
||||||
|
int? rows = null;
|
||||||
|
if (root.TryGetProperty("rowsAffected", out var ra) && ra.ValueKind == JsonValueKind.Number) rows = ra.GetInt32();
|
||||||
|
else if (root.TryGetProperty("rowsReturned", out var rr) && rr.ValueKind == JsonValueKind.Number) rows = rr.GetInt32();
|
||||||
|
if (rows is { } n && n >= 0)
|
||||||
|
return $"{op} · {n} {(n == 1 ? "row" : "rows")}";
|
||||||
|
return op;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (JsonException)
|
||||||
|
{
|
||||||
|
// Fall through — an unparseable Extra never fails projection.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return string.Equals(d.Kind, nameof(AuditKind.DbWrite), StringComparison.Ordinal)
|
||||||
|
? d.Status
|
||||||
|
: $"{d.Status} · cached";
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,4 +1,6 @@
|
|||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Tests.Debugging;
|
namespace ZB.MOM.WW.ScadaBridge.Commons.Tests.Debugging;
|
||||||
|
|
||||||
@@ -47,4 +49,103 @@ public class FlowStepProjectorTests
|
|||||||
Assert.True((edge with { Async = true }).Async);
|
Assert.True((edge with { Async = true }).Async);
|
||||||
Assert.True(new FlowEdge(a, b, false, false, Async: false, Spawn: true).Spawn);
|
Assert.True(new FlowEdge(a, b, false, false, Async: false, Spawn: true).Spawn);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- Task 1: outbound rows project onto the External lane ---------------------------------
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void SyncApiCall_Delivered_IsApiCall_External_Ok_SyncSubtitle()
|
||||||
|
{
|
||||||
|
var evt = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiOutbound, AuditKind.ApiCall, AuditStatus.Delivered,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, actor: "AckRecipe", target: "Delmia.AckRecipe", sourceNode: "node-a",
|
||||||
|
executionId: Guid.NewGuid(), sourceSiteId: "site-a", sourceInstanceId: "M1", durationMs: 212,
|
||||||
|
requestSummary: "{\"id\":1}", responseSummary: "{\"ok\":true}");
|
||||||
|
|
||||||
|
var step = FlowStepProjector.Project(evt);
|
||||||
|
|
||||||
|
Assert.Equal(FlowStepKind.ApiCall, step.Kind);
|
||||||
|
Assert.Equal(FlowLane.External, step.Lane);
|
||||||
|
Assert.Equal(FlowStepOutcome.Ok, step.Outcome);
|
||||||
|
Assert.Equal("Delmia.AckRecipe", step.Title);
|
||||||
|
Assert.Equal("Delivered · sync", step.Subtitle);
|
||||||
|
Assert.Equal("{\"id\":1}", step.InputJson);
|
||||||
|
Assert.Equal("{\"ok\":true}", step.OutputJson);
|
||||||
|
Assert.Equal(212, step.DurationMs);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void CachedSubmit_InDbOutbound_IsDbCall_Pending_CachedSubtitle()
|
||||||
|
{
|
||||||
|
var evt = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.CachedSubmit, AuditStatus.Submitted,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "MesDb", correlationId: Guid.NewGuid(), executionId: Guid.NewGuid());
|
||||||
|
|
||||||
|
var step = FlowStepProjector.Project(evt);
|
||||||
|
|
||||||
|
Assert.Equal(FlowStepKind.DbCall, step.Kind);
|
||||||
|
Assert.Equal(FlowLane.External, step.Lane);
|
||||||
|
Assert.Equal(FlowStepOutcome.Pending, step.Outcome);
|
||||||
|
Assert.Equal("Submitted · cached", step.Subtitle);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void DbWrite_WithRowsAffectedExtra_SubtitleIsOpAndRows()
|
||||||
|
{
|
||||||
|
var evt = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "MesDb", executionId: Guid.NewGuid(),
|
||||||
|
extra: "{\"op\":\"write\",\"rowsAffected\":3}");
|
||||||
|
|
||||||
|
var step = FlowStepProjector.Project(evt);
|
||||||
|
|
||||||
|
Assert.Equal(FlowStepKind.DbCall, step.Kind);
|
||||||
|
Assert.Equal(FlowStepOutcome.Ok, step.Outcome);
|
||||||
|
Assert.Equal("write · 3 rows", step.Subtitle);
|
||||||
|
|
||||||
|
var read = ScadaBridgeAuditEventFactory.Create(AuditChannel.DbOutbound, AuditKind.DbWrite, AuditStatus.Delivered,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "MesDb", extra: "{\"op\":\"read\",\"rowsReturned\":1}");
|
||||||
|
Assert.Equal("read · 1 row", FlowStepProjector.Project(read).Subtitle);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void NotifySend_Submitted_IsNotify_Pending()
|
||||||
|
{
|
||||||
|
var evt = ScadaBridgeAuditEventFactory.Create(AuditChannel.Notification, AuditKind.NotifySend, AuditStatus.Submitted,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "Planners", correlationId: Guid.NewGuid(), executionId: Guid.NewGuid());
|
||||||
|
|
||||||
|
var step = FlowStepProjector.Project(evt);
|
||||||
|
|
||||||
|
Assert.Equal(FlowStepKind.Notify, step.Kind);
|
||||||
|
Assert.Equal(FlowLane.External, step.Lane);
|
||||||
|
Assert.Equal(FlowStepOutcome.Pending, step.Outcome);
|
||||||
|
Assert.Equal("Planners", step.Title);
|
||||||
|
Assert.Equal("Submitted", step.Subtitle);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData(AuditStatus.Parked, FlowStepOutcome.Parked)]
|
||||||
|
[InlineData(AuditStatus.Discarded, FlowStepOutcome.Parked)]
|
||||||
|
[InlineData(AuditStatus.Failed, FlowStepOutcome.Failed)]
|
||||||
|
[InlineData(AuditStatus.Forwarded, FlowStepOutcome.Pending)]
|
||||||
|
[InlineData(AuditStatus.Attempted, FlowStepOutcome.Pending)]
|
||||||
|
[InlineData(AuditStatus.Delivered, FlowStepOutcome.Ok)]
|
||||||
|
public void OutboundStatus_MapsToOutcome(AuditStatus status, FlowStepOutcome expected)
|
||||||
|
{
|
||||||
|
var evt = ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiOutbound, AuditKind.CachedResolve, status,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "Delmia.AckRecipe", correlationId: Guid.NewGuid(),
|
||||||
|
errorMessage: status == AuditStatus.Failed ? "boom" : null);
|
||||||
|
|
||||||
|
var step = FlowStepProjector.Project(evt);
|
||||||
|
|
||||||
|
Assert.Equal(expected, step.Outcome);
|
||||||
|
Assert.Equal($"{status} · cached", step.Subtitle);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void ScriptTraceOutcome_Unchanged_FailedStaysFailed_OtherStatusesOk()
|
||||||
|
{
|
||||||
|
var ok = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.AttributeWrite, AuditStatus.Delivered,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "A");
|
||||||
|
var failed = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.AttributeWrite, AuditStatus.Failed,
|
||||||
|
occurredAtUtc: T0.UtcDateTime, target: "A", errorMessage: "rejected");
|
||||||
|
Assert.Equal(FlowStepOutcome.Ok, FlowStepProjector.Project(ok).Outcome);
|
||||||
|
Assert.Equal(FlowStepOutcome.Failed, FlowStepProjector.Project(failed).Outcome);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user