feat(opcua): HistoryReadEvents projects EventType, ConditionName, SourceNode; honours the where-clause (lmxopcua#746)

HistoryReadEvents projected only EventId / SourceName / Time / ReceiveTime /
Message / Severity: a select clause asking for EventType, ConditionName or
SourceNode (what most generic alarm viewers ask for) got nulls, and the
event filter's where-clause (an OfType) was ignored.

- HistoricalEvent gains optional EventTypeName, ConditionName and AlarmId.
  SqlServer, QuestDB and Gateway event reads fill them from what their
  alarm writers already store (AlarmTypeName / AlarmName / AlarmId; the
  gateway event Type and its AlarmName / AlarmId properties).
- The node manager maps the type name to its Part 9 event type (unknown or
  non-event => AlarmConditionType), projects ConditionName, and resolves
  SourceNode to the materialised condition the alarm id names (the live
  events' SourceNode), else null.
- The where-clause is evaluated per event with the SDK content filter
  through an IFilterTarget (OfType via the type tree; other operators
  against the projected fields), after the page's cursor is taken, so
  paging is unchanged.

Tested in the node manager, over a real client session (the Client.CLI has
no `historyread --events` mode yet; filed as a follow-up), and by alarm
writer -> event read round trips on SqlServer and QuestDB.

Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22
in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-10-22 17:45:38 -04:00
parent 096a3d5b3c
commit 29665c1872
13 changed files with 503 additions and 31 deletions
+15 -1
View File
@@ -585,7 +585,21 @@ owns at least one alarm condition is already an event notifier; the server regis
HistorianGateway using that source. (Alarm-history `ReadEvents` requires the gateway running
`RuntimeDb:EventReadsEnabled=true`.) Event-field projection supports the standard
`BaseEventType` select clauses — `EventId`, `SourceName`, `Time`, `ReceiveTime`, `Message`,
and `Severity`; an unsupported select operand returns a null field (spec-conformant).
`Severity` — plus (client-visible: these used to be null):
- **`EventType`** — the stored alarm type name mapped to its Part 9 type (`LimitAlarm` ⇒ `LimitAlarmType`, `OffNormalAlarm` ⇒
`OffNormalAlarmType`, `DiscreteAlarm` ⇒ `DiscreteAlarmType`, `AlarmCondition` ⇒ `AlarmConditionType`); a row with no type
name, or one that is not an event type, is `AlarmConditionType` (every historical row is alarm history).
- **`ConditionName`** — the stored alarm name.
- **`SourceNode`** — the condition node the stored alarm id names (a native alarm's RawPath, a scripted alarm's id), the same
`SourceNode` live condition events carry; null when that condition is not materialised on this node now.
Every backend stores the three (SqlServer `AlarmTypeName` / `AlarmName` / `AlarmId`, QuestDB `alarm_type_name` /
`alarm_name` / `alarm_id`, Gateway the event `Type` and its `AlarmName` / `AlarmId` properties). **The where-clause is
honoured:** each event is evaluated against it with the SDK's content filter — `OfType` through the server's type tree (a
supertype keeps its subtypes), any other operator against the projected fields. Filtering happens after a page's
continuation point is taken from the unfiltered page, so paging is unchanged: a filtered page may hold fewer events (even
none) and still carry a continuation point. An unsupported select operand returns a null field (spec-conformant).
### No driver-side history (Gitea #655)
@@ -118,13 +118,26 @@ public sealed class HistoryReadLimitExceededException : InvalidOperationExceptio
/// <param name="ReceivedTimeUtc">Historian-side timestamp — when the historian persisted the row; may lag <paramref name="EventTimeUtc"/> by the historian's buffer flush cadence.</param>
/// <param name="Message">Human-readable message text.</param>
/// <param name="Severity">OPC UA severity (1-1000). Backends map their native priority scale onto this range.</param>
/// <param name="EventTypeName">
/// The stored alarm type name (<c>AlarmTypeName</c> — e.g. <c>AlarmCondition</c>, <c>LimitAlarm</c>,
/// <c>OffNormalAlarm</c>, <c>DiscreteAlarm</c>), or null when the backend has none. The server maps it to the OPC UA
/// event type NodeId it projects as <c>EventType</c> and filters an <c>OfType</c> where-clause by.
/// </param>
/// <param name="ConditionName">The condition's name (the stored <c>AlarmName</c>), projected as <c>ConditionName</c>.</param>
/// <param name="AlarmId">
/// The stored alarm id (a native alarm's RawPath / a scripted alarm's id) — what the server resolves the event's
/// <c>SourceNode</c> (the condition node) from, when that condition is materialised.
/// </param>
public sealed record HistoricalEvent(
string EventId,
string? SourceName,
DateTime EventTimeUtc,
DateTime ReceivedTimeUtc,
string? Message,
ushort Severity);
ushort Severity,
string? EventTypeName = null,
string? ConditionName = null,
string? AlarmId = null);
/// <summary>Result of a <see cref="IHistorianDataSource.ReadEventsAsync(string?, DateTime, DateTime, int, CancellationToken)"/> call.</summary>
/// <param name="Events">Events ordered by <c>EventTimeUtc</c> in the read's direction — ascending, or descending for a reverse read.</param>
@@ -33,9 +33,16 @@ internal static class EventMapper
EventTimeUtc: historianEvent.EventTime?.ToDateTime() ?? default, // Utc kind
ReceivedTimeUtc: historianEvent.ReceivedTime?.ToDateTime() ?? default, // Utc kind
Message: message,
Severity: ParseSeverity(historianEvent.Properties));
Severity: ParseSeverity(historianEvent.Properties),
// What GatewayAlarmHistorianWriter sent (AlarmEventMapper): Type = AlarmTypeName, AlarmName / AlarmId properties.
EventTypeName: string.IsNullOrEmpty(historianEvent.Type) ? null : historianEvent.Type,
ConditionName: Property(historianEvent.Properties, "AlarmName"),
AlarmId: Property(historianEvent.Properties, "AlarmId"));
}
private static string? Property(IDictionary<string, string> properties, string name) =>
properties.TryGetValue(name, out var v) && !string.IsNullOrEmpty(v) ? v : null;
/// <summary>Maps a batch of gateway events to historical events, in order.</summary>
/// <param name="events">The gateway wire events.</param>
/// <returns>The driver-agnostic historical events.</returns>
@@ -449,7 +449,7 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
var (lo, hi) = start <= end ? (start, end) : (end, start);
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
var sql =
$"SELECT event_id, source, ts, received_utc, message, severity FROM {_conn.EventTable} " +
$"SELECT event_id, source, ts, received_utc, message, severity, alarm_type_name, alarm_name, alarm_id FROM {_conn.EventTable} " +
(reverse ? "WHERE ts > $1 AND ts <= $2" : "WHERE ts >= $1 AND ts < $2") +
(sourceName is null ? "" : " AND source = $3") +
(reverse ? " ORDER BY ts DESC, event_id DESC" : " ORDER BY ts ASC, event_id ASC") +
@@ -474,7 +474,10 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
eventTime,
reader.IsDBNull(3) ? eventTime : QuestDbConnection.ReadUtc(reader, 3),
reader.IsDBNull(4) ? null : reader.GetString(4),
HistoricalEventRows.ClampSeverity(reader.IsDBNull(5) ? 0 : reader.GetInt32(5)));
HistoricalEventRows.ClampSeverity(reader.IsDBNull(5) ? 0 : reader.GetInt32(5)),
EventTypeName: HistoricalEventRows.NullIfEmpty(reader.IsDBNull(6) ? null : reader.GetString(6)),
ConditionName: HistoricalEventRows.NullIfEmpty(reader.IsDBNull(7) ? null : reader.GetString(7)),
AlarmId: HistoricalEventRows.NullIfEmpty(reader.IsDBNull(8) ? null : reader.GetString(8)));
if (bounded && events.Count == maxEvents)
{
continuation = Encoding.UTF8.GetBytes(EventContinuation.Encode(e.EventTimeUtc, e.EventId));
@@ -7,4 +7,9 @@ public static class HistoricalEventRows
/// <param name="severity">Stored severity.</param>
/// <returns>The severity clamped to 0..65535.</returns>
public static ushort ClampSeverity(int severity) => (ushort)Math.Clamp(severity, 0, ushort.MaxValue);
/// <summary>A stored optional text column as the event carries it: empty (the writers' "none") is null.</summary>
/// <param name="value">The stored text.</param>
/// <returns>The text, or null when empty or absent.</returns>
public static string? NullIfEmpty(string? value) => string.IsNullOrEmpty(value) ? null : value;
}
@@ -396,7 +396,7 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc);
var bounded = maxEvents > 0 && maxEvents < int.MaxValue;
var sql =
$"SELECT {(bounded ? "TOP (@take) " : "")}EventId, SourceName, EventTimeUtc, ReceivedTimeUtc, Message, Severity " +
$"SELECT {(bounded ? "TOP (@take) " : "")}EventId, SourceName, EventTimeUtc, ReceivedTimeUtc, Message, Severity, AlarmTypeName, AlarmName, AlarmId " +
$"FROM {_conn.Q("Event")} WHERE " +
(reverse ? "EventTimeUtc > @lo AND EventTimeUtc <= @hi" : "EventTimeUtc >= @lo AND EventTimeUtc < @hi") +
(sourceName is null ? "" : " AND SourceName = @source") +
@@ -421,7 +421,10 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource
SqlHistorianConnection.GetUtc(reader, 2),
SqlHistorianConnection.GetUtc(reader, 3),
reader.IsDBNull(4) ? null : reader.GetString(4),
HistoricalEventRows.ClampSeverity(reader.GetInt32(5)));
HistoricalEventRows.ClampSeverity(reader.GetInt32(5)),
EventTypeName: HistoricalEventRows.NullIfEmpty(reader.IsDBNull(6) ? null : reader.GetString(6)),
ConditionName: HistoricalEventRows.NullIfEmpty(reader.IsDBNull(7) ? null : reader.GetString(7)),
AlarmId: HistoricalEventRows.NullIfEmpty(reader.IsDBNull(8) ? null : reader.GetString(8)));
if (bounded && events.Count == maxEvents)
{
continuation = Encoding.UTF8.GetBytes(EventContinuation.Encode(e.EventTimeUtc, e.EventId));
@@ -5714,8 +5714,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
List<NodeHandle> nodesToProcess,
IDictionary<NodeId, NodeState> cache)
{
// Snapshot the select clauses once — the same filter projects every node's events.
// Snapshot the select clauses once — the same filter projects every node's events — and the where-clause, which
// is evaluated per event against the same projection (an OfType through the server's type tree).
var selectClauses = details.Filter?.SelectClauses ?? new SimpleAttributeOperandCollection();
var where = details.Filter?.WhereClause is { Elements.Count: > 0 } w ? w : null;
var filterContext = where is null ? null : new FilterContext(Server.NamespaceUris, Server.TypeTree, context.PreferredLocales, Server.Telemetry);
var session = context.OperationContext?.Session;
nodesToProcess = AuthorizeHistoryRead(context, nodesToProcess, results, errors);
@@ -5739,7 +5742,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
}
var sn = sourceName;
work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, nodesToRead, session, results, errors, ct));
work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, where, filterContext, nodesToRead, session,
results, errors, ct));
}
RunBounded(work, HistoryReadBatchParallelism);
@@ -5757,6 +5761,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <param name="sourceName">The registered historian event-source name for this notifier.</param>
/// <param name="details">The event-read details (window + count cap).</param>
/// <param name="selectClauses">The event-filter select clauses (the fields to emit, in order).</param>
/// <param name="where">The event filter's where-clause, or null for none.</param>
/// <param name="filterContext">The context the where-clause is evaluated in (null when there is none).</param>
/// <param name="results">The service-level results list to fill at <c>handle.Index</c>.</param>
/// <param name="errors">The service-level errors list to fill at <c>handle.Index</c>.</param>
/// <param name="ct">Per-request deadline token.</param>
@@ -5765,6 +5771,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
string? sourceName,
ReadEventDetails details,
SimpleAttributeOperandCollection selectClauses,
ContentFilter? where,
FilterContext? filterContext,
IList<HistoryReadValueId> nodesToRead,
ISession? session,
IList<SdkHistoryReadResult> results,
@@ -5830,7 +5838,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
results[handle.Index] = new SdkHistoryReadResult
{
StatusCode = StatusCodes.GoodNoData,
HistoryData = new ExtensionObject(ProjectEvents([], selectClauses)),
HistoryData = new ExtensionObject(ProjectEvents([], selectClauses, null, null)),
};
errors[handle.Index] = ServiceResult.Good;
return;
@@ -5961,12 +5969,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
EmptyToTicks: walkHint.EmptyToTicks);
}
var historyEvent = ProjectEvents(events, selectClauses);
// The where-clause filters AFTER the page's cursor was taken from the unfiltered events, so paging (time,
// boundary skip) is exactly as without a filter: a page may come back with fewer events — even none — and a
// continuation point, never skipping or repeating one.
var historyEvent = ProjectEvents(events, selectClauses, where, filterContext);
var outboundCp = pendingCursor is null ? null : _historyContinuationStore.Save(session, pendingCursor);
results[handle.Index] = new SdkHistoryReadResult
{
// No events ⇒ GoodNoData (the notifier is historized, the window just held no events).
StatusCode = events.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good,
// No events ⇒ GoodNoData (the notifier is historized, the window just held no events — or none matched).
StatusCode = historyEvent.Events.Count == 0 && outboundCp is null ? StatusCodes.GoodNoData : StatusCodes.Good,
HistoryData = new ExtensionObject(historyEvent),
// Null when the page was complete, or when no session is available to hold the cursor.
ContinuationPoint = outboundCp,
@@ -6415,23 +6426,28 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <summary>
/// Project a sequence of <see cref="HistoricalEvent"/>s into an SDK <see cref="HistoryEvent"/> —
/// one <see cref="HistoryEventFieldList"/> per event, each carrying the requested
/// <paramref name="selectClauses"/>' fields in select-clause order (see
/// <see cref="ProjectEventField"/>).
/// one <see cref="HistoryEventFieldList"/> per event that passes <paramref name="where"/>, each carrying the
/// requested <paramref name="selectClauses"/>' fields in select-clause order (see <see cref="ProjectEventField"/>).
/// </summary>
/// <param name="events">The historian's event rows.</param>
/// <param name="selectClauses">The request's event-filter select clauses (the fields to emit, in order).</param>
/// <param name="where">The request's where-clause, or null for none (every event passes).</param>
/// <param name="filterContext">The context <paramref name="where"/> is evaluated in.</param>
/// <returns>The populated SDK <see cref="HistoryEvent"/>.</returns>
private static HistoryEvent ProjectEvents(
IReadOnlyList<HistoricalEvent> events, SimpleAttributeOperandCollection selectClauses)
private HistoryEvent ProjectEvents(
IReadOnlyList<HistoricalEvent> events, SimpleAttributeOperandCollection selectClauses,
ContentFilter? where, FilterContext? filterContext)
{
var fieldLists = new HistoryEventFieldListCollection(events.Count);
foreach (var evt in events)
{
var target = new HistoricalEventFilterTarget(this, evt);
if (where is not null && filterContext is not null && !where.Evaluate(filterContext, target)) continue;
var fields = new VariantCollection(selectClauses.Count);
foreach (var operand in selectClauses)
{
fields.Add(ProjectEventField(evt, operand));
fields.Add(target.Project(operand));
}
fieldLists.Add(new HistoryEventFieldList { EventFields = fields });
@@ -6440,32 +6456,98 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
return new HistoryEvent { Events = fieldLists };
}
/// <summary>
/// A historical event as the SDK's content-filter evaluator sees it: its type (for <c>OfType</c>, through the
/// server's type tree) and its projected fields (for every other operator).
/// </summary>
private sealed class HistoricalEventFilterTarget(OtOpcUaNodeManager owner, HistoricalEvent evt) : IFilterTarget
{
private NodeId? _type;
/// <summary>The event's OPC UA event type (see <see cref="OtOpcUaNodeManager.EventTypeOf"/>).</summary>
public NodeId EventType => _type ??= owner.EventTypeOf(evt.EventTypeName);
public Variant Project(SimpleAttributeOperand operand) => owner.ProjectEventField(evt, operand, EventType);
public bool IsTypeOf(IFilterContext context, NodeId typeDefinitionId) =>
EventType == typeDefinitionId || context.TypeTree.IsTypeOf(EventType, typeDefinitionId);
public object? GetAttributeValue(IFilterContext context, NodeId typeDefinitionId, IList<QualifiedName> relativePath,
uint attributeId, NumericRange indexRange)
{
if (attributeId != Attributes.Value || relativePath is null || relativePath.Count == 0) return null;
var operand = new SimpleAttributeOperand { TypeDefinitionId = typeDefinitionId, AttributeId = attributeId };
foreach (var name in relativePath) operand.BrowsePath.Add(name);
return Project(operand).Value;
}
}
/// <summary>
/// The OPC UA event type of a historical alarm event, from its stored alarm type name: the Part 9 type of that name
/// (<c>LimitAlarm</c> ⇒ <c>LimitAlarmType</c>, <c>OffNormalAlarm</c> ⇒ <c>OffNormalAlarmType</c>, …, with or without the
/// <c>Type</c> suffix) when it is an event type in the server's type tree; otherwise — no name, or a name that is not
/// an event type — <c>AlarmConditionType</c>: every historical row is alarm history, of a subtype not modelled.
/// </summary>
/// <param name="typeName">The stored alarm type name, or null.</param>
/// <returns>The event type NodeId.</returns>
private NodeId EventTypeOf(string? typeName)
{
if (string.IsNullOrWhiteSpace(typeName)) return ObjectTypeIds.AlarmConditionType;
return s_eventTypesByName.GetOrAdd(typeName, name =>
{
var field = typeof(ObjectTypeIds).GetField(name.EndsWith("Type", StringComparison.Ordinal) ? name : name + "Type",
System.Reflection.BindingFlags.Public | System.Reflection.BindingFlags.Static);
return field?.GetValue(null) is NodeId type && Server.TypeTree.IsTypeOf(type, ObjectTypeIds.BaseEventType)
? type
: ObjectTypeIds.AlarmConditionType;
});
}
private static readonly ConcurrentDictionary<string, NodeId> s_eventTypesByName = new(StringComparer.Ordinal);
/// <summary>
/// The <c>SourceNode</c> of a historical alarm event: the materialised condition its alarm id names (a native
/// condition's id is its RawPath, Raw realm; a scripted one its alarm id, UNS realm) — the SourceNode the live
/// condition events carry. Null when no such condition is materialised now (removed, or never on this node).
/// </summary>
private NodeId? SourceNodeOf(HistoricalEvent evt)
{
if (string.IsNullOrEmpty(evt.AlarmId)) return null;
foreach (var realm in new[] { AddressSpaceRealm.Raw, AddressSpaceRealm.Uns })
if (_alarmConditions.TryGetValue(MapKey(realm, evt.AlarmId), out var condition)) return condition.NodeId;
return null;
}
/// <summary>
/// Project one <see cref="HistoricalEvent"/> field requested by a select <paramref name="operand"/>
/// into a <see cref="Variant"/>. Mapping is by the operand's BrowsePath LEAF QualifiedName name
/// (case-sensitive, per the OPC UA BaseEventType field names) against the
/// (case-sensitive, per the OPC UA BaseEventType / ConditionType field names) against the
/// <see cref="HistoricalEvent"/> shape:
/// <c>EventId</c>→ByteString, <c>SourceName</c>→String, <c>Time</c>→DateTime,
/// <c>ReceiveTime</c>→DateTime, <c>Message</c>→LocalizedText, <c>Severity</c>→UInt16. Any other
/// leaf (EventType / SourceNode / ConditionName / an unrecognised name) and an empty BrowsePath ⇒
/// <see cref="Variant.Null"/> — spec-conformant: a field the server can't supply is null.
/// <c>EventId</c>→ByteString, <c>EventType</c>→NodeId (<paramref name="eventType"/>), <c>SourceName</c>→String,
/// <c>SourceNode</c>→NodeId (the resolved condition, else null), <c>Time</c>/<c>ReceiveTime</c>→DateTime,
/// <c>Message</c>→LocalizedText, <c>Severity</c>→UInt16, <c>ConditionName</c>→String. Any other leaf and an empty
/// BrowsePath ⇒ <see cref="Variant.Null"/> — spec-conformant: a field the server can't supply is null.
/// </summary>
/// <param name="evt">The source event row.</param>
/// <param name="operand">The select-clause operand naming the field to project.</param>
/// <param name="eventType">The event's OPC UA event type (<see cref="EventTypeOf"/>).</param>
/// <returns>The projected variant, or <see cref="Variant.Null"/> for an unsupported field.</returns>
private static Variant ProjectEventField(HistoricalEvent evt, SimpleAttributeOperand operand)
private Variant ProjectEventField(HistoricalEvent evt, SimpleAttributeOperand operand, NodeId eventType)
{
var leaf = LeafFieldName(operand);
return leaf switch
{
// BaseEventType/EventId is a ByteString — encode the driver-specific string id as UTF-8 bytes.
"EventId" => new Variant(System.Text.Encoding.UTF8.GetBytes(evt.EventId ?? string.Empty)),
"EventType" => new Variant(eventType),
"SourceName" => evt.SourceName is null ? Variant.Null : new Variant(evt.SourceName),
"SourceNode" => SourceNodeOf(evt) is { } source ? new Variant(source) : Variant.Null,
"Time" => new Variant(evt.EventTimeUtc),
"ReceiveTime" => new Variant(evt.ReceivedTimeUtc),
"Message" => new Variant(new LocalizedText(evt.Message ?? string.Empty)),
"Severity" => new Variant(evt.Severity), // UInt16
// EventType / SourceNode / ConditionName / empty path / unrecognised leaf ⇒ null (spec-conformant).
"ConditionName" => evt.ConditionName is null ? Variant.Null : new Variant(evt.ConditionName),
// empty path / unrecognised leaf ⇒ null (spec-conformant).
_ => Variant.Null,
};
}
@@ -7,6 +7,26 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.Mapping;
public sealed class EventMapperTests
{
[Fact]
public void Carries_the_alarm_type_condition_name_and_alarm_id_the_writer_sent()
{
// The round trip of AlarmEventMapper's shape: Type = AlarmTypeName, Properties AlarmName / AlarmId.
var sent = AlarmEventMapper.ToHistorianEvent(new ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.AlarmHistorianEvent(
"Plant/Modbus/dev1/temp_hi", "Area/Line/Eq", "HiTemp", "LimitAlarm", ZB.MOM.WW.OtOpcUa.Core.Abstractions.AlarmSeverity.High,
"Activated", "high", "u", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc)));
var h = EventMapper.ToHistoricalEvent(sent);
Assert.Equal("LimitAlarm", h.EventTypeName);
Assert.Equal("HiTemp", h.ConditionName);
Assert.Equal("Plant/Modbus/dev1/temp_hi", h.AlarmId);
// A gateway row without them (e.g. an event another writer produced) maps to nulls, never throws.
var bare = new HistorianEvent { Id = "E", EventTime = Ts(2026, 1, 1, 0, 0, 0), ReceivedTime = Ts(2026, 1, 1, 0, 0, 0) };
var b = EventMapper.ToHistoricalEvent(bare);
Assert.Null(b.EventTypeName);
Assert.Null(b.ConditionName);
Assert.Null(b.AlarmId);
}
[Fact]
public void Maps_core_fields_and_times()
{
@@ -32,6 +32,25 @@ public sealed class AlarmWriterTests(QuestDbFixture fx)
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
}
[Fact]
public async Task An_event_reads_back_with_its_type_name_condition_name_and_alarm_id()
{
fx.SkipIfUnavailable();
var o = fx.Options(fx.UniquePrefix());
var c = new QuestDbConnection(o);
var boot = new QuestDbSchemaBootstrapper(c, o);
var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
var w = new QuestDbAlarmHistorianWriter(c, boot, wal);
(await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
var src = new QuestDbHistorianDataSource(c, boot, new QuestDbTagCatalog(c, boot, wal));
var e = (await src.ReadEventsAsync("Plant/Line/Eq", T.AddMinutes(-1), T.AddMinutes(1), 0, CancellationToken.None))
.Events.ShouldHaveSingleItem();
e.EventTypeName.ShouldBe("LimitAlarm");
e.ConditionName.ShouldBe("HiTemp");
e.AlarmId.ShouldBe("alarm-1");
}
[Fact]
public async Task A_batch_is_one_insert_per_chunk_and_a_redelivery_converges()
{
@@ -63,6 +63,25 @@ public sealed class AlarmWriterTests(SqlHistorianFixture fx)
r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]);
}
[Fact]
public async Task An_event_reads_back_with_its_type_name_condition_name_and_alarm_id()
{
fx.SkipIfUnavailable();
var s = SqlHistorianFixture.UniqueSchema();
var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s };
var c = new SqlHistorianConnection(o);
var boot = new SqlServerSchemaBootstrapper(c, o);
var w = new SqlServerAlarmHistorianWriter(c, boot);
(await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
using var src = new SqlServerHistorianDataSource(c, boot, new SqlServerTagCatalog(c, boot));
var e = (await src.ReadEventsAsync("Plant/Line/Eq", T.AddMinutes(-1), T.AddMinutes(1), 0, CancellationToken.None))
.Events.ShouldHaveSingleItem();
e.EventTypeName.ShouldBe("LimitAlarm");
e.ConditionName.ShouldBe("HiTemp");
e.AlarmId.ShouldBe("alarm-1");
}
[Fact]
public async Task A_batch_is_one_statement_per_chunk_and_a_redelivery_converges()
{
@@ -0,0 +1,107 @@
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using CoreHistoryReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Over a real SDK client session: a HistoryReadEvents whose select clause includes <c>EventType</c>,
/// <c>ConditionName</c> and <c>SourceNode</c> — what a generic alarm viewer asks for — gets non-null values (they all
/// used to be null), and an <c>OfType</c> where-clause is honoured on the wire.
/// </summary>
public sealed class EventHistoryProjectionOverTheWireTests
{
private static readonly DateTime Epoch = new(2026, 10, 22, 0, 0, 0, DateTimeKind.Utc);
[Fact]
public async Task EventType_ConditionName_and_SourceNode_come_back_non_null_and_OfType_filters()
{
var ct = TestContext.Current.CancellationToken;
var rig = await StartAsync("EvtProjectionWire", encrypted: false, ct);
try
{
rig.Nm.HistorianDataSource = new FixedSource(
[
new HistoricalEvent("e-limit", RawDeviceFolder, Epoch, Epoch, "high", 700,
EventTypeName: "LimitAlarm", ConditionName: "temp_hi", AlarmId: RawAlarmPath),
new HistoricalEvent("e-off", RawDeviceFolder, Epoch.AddSeconds(1), Epoch, "door", 500,
EventTypeName: "OffNormalAlarm", ConditionName: "door_open", AlarmId: RawAlarmPath),
]);
rig.MaterialiseNative();
rig.Enforce(false, [Key()]);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var node = new NodeId(RawDeviceFolder, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
var all = await ReadAsync(session, node, where: null, ct);
all.Count.ShouldBe(2);
foreach (var fields in all)
{
fields[1].Value.ShouldNotBeNull("EventType");
fields[2].Value.ShouldNotBeNull("ConditionName");
fields[3].Value.ShouldNotBeNull("SourceNode");
}
all[0][1].Value.ShouldBe(ObjectTypeIds.LimitAlarmType);
all[0][2].Value.ShouldBe("temp_hi");
all[0][3].Value.ShouldBe(ConditionNodeId(session, native: true), "the native condition the alarm id names");
var where = new ContentFilter();
where.Push(FilterOperator.OfType, new LiteralOperand(new Variant(ObjectTypeIds.LimitAlarmType)));
var limits = await ReadAsync(session, node, where, ct);
limits.Select(f => System.Text.Encoding.UTF8.GetString((byte[])f[0].Value)).ShouldBe(["e-limit"]);
}
finally { await rig.DisposeAsync(); }
}
private static async Task<List<VariantCollection>> ReadAsync(ISession session, NodeId node, ContentFilter? where, CancellationToken ct)
{
var filter = new EventFilter { WhereClause = where ?? new ContentFilter() };
foreach (var leaf in new[] { "EventId", "EventType", "ConditionName", "SourceNode" })
{
var type = leaf == "ConditionName" ? ObjectTypeIds.ConditionType : ObjectTypeIds.BaseEventType;
filter.SelectClauses.Add(new SimpleAttributeOperand(type, new QualifiedName(leaf)) { AttributeId = Attributes.Value });
}
var details = new ReadEventDetails
{
StartTime = Epoch.AddDays(-1),
EndTime = Epoch.AddDays(1),
NumValuesPerNode = 100,
Filter = filter,
};
var response = await session.HistoryReadAsync(
null, new ExtensionObject(details), TimestampsToReturn.Both, false,
new HistoryReadValueIdCollection { new HistoryReadValueId { NodeId = node } }, ct);
var result = response.Results.ShouldHaveSingleItem();
StatusCode.IsBad(result.StatusCode).ShouldBeFalse($"HistoryRead failed: {result.StatusCode}");
var he = (HistoryEvent)ExtensionObject.ToEncodeable(result.HistoryData);
return he.Events.Select(e => e.EventFields).ToList();
}
private sealed class FixedSource(IReadOnlyList<HistoricalEvent> events) : IHistorianDataSource
{
public Task<HistoricalEventsResult> ReadEventsAsync(string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken ct) =>
Task.FromResult(new HistoricalEventsResult(events.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList(), null));
public Task<CoreHistoryReadResult> ReadRawAsync(string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
CancellationToken cancellationToken) =>
NullHistorianDataSource.Instance.ReadRawAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken);
public Task<CoreHistoryReadResult> ReadProcessedAsync(string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
public Task<CoreHistoryReadResult> ReadAtTimeAsync(string fullReference, IReadOnlyList<DateTime> timestampsUtc,
CancellationToken cancellationToken) =>
NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
}
@@ -0,0 +1,179 @@
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// HistoryReadEvents projects every field a historical alarm event can supply — <c>EventType</c> (the stored alarm
/// type name mapped to its Part 9 type, an unmodelled one to <c>AlarmConditionType</c>), <c>ConditionName</c>, and
/// <c>SourceNode</c> when the event's alarm id resolves to a materialised condition — and honours the request's
/// where-clause (an <c>OfType</c> by the type tree; any other operator against the projected fields). These used to
/// project <see cref="Variant.Null"/> and the where-clause was ignored.
/// </summary>
public sealed class NodeManagerHistoryReadEventsProjectionTests : IDisposable
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static readonly DateTime T0 = new(2026, 10, 22, 8, 0, 0, DateTimeKind.Utc);
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-evtproj-{Guid.NewGuid():N}");
private const string Equipment = "eq-proj";
private static readonly HistoricalEvent[] Events =
[
new("e-limit", Equipment, T0, T0, "hi", 700, EventTypeName: "LimitAlarm", ConditionName: "HighTemp", AlarmId: "alarm-1"),
new("e-offnormal", Equipment, T0.AddSeconds(1), T0, "off", 500, EventTypeName: "OffNormalAlarm", ConditionName: "Door", AlarmId: "alarm-gone"),
new("e-untyped", Equipment, T0.AddSeconds(2), T0, "x", 300),
new("e-bogus", Equipment, T0.AddSeconds(3), T0, "y", 900, EventTypeName: "Folder", ConditionName: "Odd"),
];
[Fact]
public async Task EventType_ConditionName_and_a_resolvable_SourceNode_are_projected()
{
var (host, server, nm, notifier) = await BootAsync();
var condition = nm.TryGetAlarmCondition("alarm-1")!.NodeId;
var history = Read(server, nm, notifier, Select("EventId", "EventType", "ConditionName", "SourceNode"), where: null);
var rows = history.Events.ToDictionary(e => System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value));
rows["e-limit"].EventFields[1].Value.ShouldBe(ObjectTypeIds.LimitAlarmType);
rows["e-limit"].EventFields[2].Value.ShouldBe("HighTemp");
rows["e-limit"].EventFields[3].Value.ShouldBe(condition, "the alarm id resolves to the materialised condition — the live events' SourceNode");
rows["e-offnormal"].EventFields[1].Value.ShouldBe(ObjectTypeIds.OffNormalAlarmType);
rows["e-offnormal"].EventFields[3].Value.ShouldBeNull("no condition is materialised for that alarm id");
// No stored type: still an alarm condition (every historical row is alarm history); an unrecognised or non-event
// type name is not passed through as a non-event type.
rows["e-untyped"].EventFields[1].Value.ShouldBe(ObjectTypeIds.AlarmConditionType);
rows["e-untyped"].EventFields[2].Value.ShouldBeNull();
rows["e-bogus"].EventFields[1].Value.ShouldBe(ObjectTypeIds.AlarmConditionType);
await host.DisposeAsync();
}
[Fact]
public async Task An_OfType_where_clause_keeps_only_events_of_that_type_or_a_subtype()
{
var (host, server, nm, notifier) = await BootAsync();
Ids(Read(server, nm, notifier, Select("EventId"), OfType(ObjectTypeIds.LimitAlarmType))).ShouldBe(["e-limit"]);
Ids(Read(server, nm, notifier, Select("EventId"), OfType(ObjectTypeIds.OffNormalAlarmType))).ShouldBe(["e-offnormal"]);
// A supertype keeps its subtypes: every row is an AlarmConditionType (positive control — the filter is not
// simply dropping everything typed).
Ids(Read(server, nm, notifier, Select("EventId"), OfType(ObjectTypeIds.AlarmConditionType)))
.ShouldBe(["e-limit", "e-offnormal", "e-untyped", "e-bogus"]);
// An unrelated event type keeps nothing (GoodNoData).
Ids(Read(server, nm, notifier, Select("EventId"), OfType(ObjectTypeIds.AuditEventType))).ShouldBeEmpty();
await host.DisposeAsync();
}
[Fact]
public async Task A_field_where_clause_is_evaluated_against_the_projected_fields()
{
var (host, server, nm, notifier) = await BootAsync();
var where = new ContentFilter();
where.Push(FilterOperator.GreaterThan,
new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("Severity")) { AttributeId = Attributes.Value },
new LiteralOperand(new Variant((ushort)600)));
Ids(Read(server, nm, notifier, Select("EventId"), where)).ShouldBe(["e-limit", "e-bogus"]);
await host.DisposeAsync();
}
private static string[] Ids(HistoryEvent h) =>
h.Events.Select(e => System.Text.Encoding.UTF8.GetString((byte[])e.EventFields[0].Value)).ToArray();
private static ContentFilter OfType(NodeId type)
{
var f = new ContentFilter();
f.Push(FilterOperator.OfType, new LiteralOperand(new Variant(type)));
return f;
}
private static SimpleAttributeOperandCollection Select(params string[] leaves)
{
var c = new SimpleAttributeOperandCollection();
foreach (var leaf in leaves)
c.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName(leaf)) { AttributeId = Attributes.Value });
return c;
}
private static HistoryEvent Read(OtOpcUaSdkServer server, OtOpcUaNodeManager nm, NodeId notifier,
SimpleAttributeOperandCollection select, ContentFilter? where)
{
var details = new ReadEventDetails
{
StartTime = T0.AddHours(-1),
EndTime = T0.AddHours(1),
NumValuesPerNode = 100,
Filter = new EventFilter { SelectClauses = select, WhereClause = where ?? new ContentFilter() },
};
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
var results = new List<SdkHistoryReadResult> { null! };
var errors = new List<ServiceResult> { null! };
nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false,
new List<HistoryReadValueId> { new() { NodeId = notifier } }, results, errors);
StatusCode.IsGood(errors[0].StatusCode).ShouldBeTrue($"HistoryRead failed: {errors[0].StatusCode}");
return (HistoryEvent)ExtensionObject.ToEncodeable(results[0].HistoryData);
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server, OtOpcUaNodeManager Nm, NodeId Notifier)> BootAsync()
{
var started = await OpcUaHostStart.StartAsync(
port => new OpcUaApplicationHost(
new OpcUaApplicationHostOptions
{
ApplicationName = "OtOpcUa.HistoryReadEventsProjectionTest",
ApplicationUri = $"urn:OtOpcUa.HistoryReadEventsProjectionTest:{Guid.NewGuid():N}",
OpcUaPort = port,
PublicHostname = "localhost",
PkiStoreRoot = _pkiRoot,
},
NullLogger<OpcUaApplicationHost>.Instance),
() => new OtOpcUaSdkServer(), Ct);
var nm = started.Server.NodeManager!;
nm.HistorianDataSource = new FixedEventsSource(Events);
nm.EnsureFolder(Equipment, parentNodeId: null, displayName: "Equipment", realm: AddressSpaceRealm.Uns);
nm.MaterialiseAlarmCondition("alarm-1", Equipment, "HighTemp", "LimitAlarm", severity: 700, realm: AddressSpaceRealm.Uns);
return (started.Host, started.Server, nm, nm.TryGetFolder(Equipment)!.NodeId);
}
private sealed class FixedEventsSource(IReadOnlyList<HistoricalEvent> events) : IHistorianDataSource
{
public Task<HistorianRead> ReadRawAsync(string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
CancellationToken cancellationToken) =>
NullHistorianDataSource.Instance.ReadRawAsync(fullReference, startUtc, endUtc, maxValuesPerNode, cancellationToken);
public Task<HistorianRead> ReadProcessedAsync(string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
public Task<HistorianRead> ReadAtTimeAsync(string fullReference, IReadOnlyList<DateTime> timestampsUtc,
CancellationToken cancellationToken) =>
NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
public Task<HistoricalEventsResult> ReadEventsAsync(string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken cancellationToken) =>
Task.FromResult(new HistoricalEventsResult(events.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList(), null));
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
public void Dispose()
{
if (Directory.Exists(_pkiRoot))
{
try { Directory.Delete(_pkiRoot, recursive: true); }
catch { /* best-effort cleanup */ }
}
}
}
@@ -133,8 +133,9 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
await host.DisposeAsync();
}
/// <summary>An unsupported select operand (BrowsePath ["EventType"]) projects to Variant.Null — a field
/// the server can't supply is null (spec-conformant) — while supported siblings still project.</summary>
/// <summary>An unsupported select operand (BrowsePath ["LocalTime"]) projects to Variant.Null — a field
/// the server can't supply is null (spec-conformant) — while supported siblings still project. (EventType used to
/// be such a field; it is projected now — see NodeManagerHistoryReadEventsProjectionTests.)</summary>
[Fact]
public async Task Events_unsupported_select_field_projects_null()
{
@@ -156,8 +157,8 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
StartTime = DateTime.UtcNow.AddHours(-1),
EndTime = DateTime.UtcNow,
NumValuesPerNode = 10,
// EventType is a real BaseEventType field we cannot supply from a HistoricalEvent; EventId is.
Filter = SelectFilter("EventType", "EventId"),
// LocalTime is a real BaseEventType field we cannot supply from a HistoricalEvent; EventId is.
Filter = SelectFilter("LocalTime", "EventId"),
};
var (results, errors) = InvokeHistoryRead(server, nm, details, notifierNodeId);
@@ -166,7 +167,7 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
var history = (HistoryEvent)ExtensionObject.ToEncodeable(results[0].HistoryData);
var fields = history.Events[0].EventFields;
fields.Count.ShouldBe(2);
// EventType ⇒ Variant.Null.
// LocalTime ⇒ Variant.Null.
fields[0].Value.ShouldBeNull();
// EventId still projects to a ByteString.
fields[1].Value.ShouldBeOfType<byte[]>().ShouldBe(System.Text.Encoding.UTF8.GetBytes("evt-1"));