dohertj2/lmxopcua
1
0
Fork 0
Code Issues 12 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(mesh-phase4): LocalDb alarm-condition-state store (replicated, pair-local)

Browse Source 
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-23 12:19:44 -04:00
parent 34b613d942
commit 4f4cdd05ec
7 changed files with 754 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs
+82
View File
@@ -0,0 +1,82 @@
using Microsoft.Data.Sqlite;
namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
/// <summary>
/// DDL for the Part 9 scripted-alarm condition state: one row per alarm identity holding the
/// operator-supplied state (Enabled / Acked / Confirmed / Shelving) plus the ack/confirm audit
/// trail and comment history.
/// </summary>
/// <remarks>
/// <para>
/// Replaces the central config DB's <c>ScriptedAlarmState</c> table as the node-local home
/// for this state (per-cluster mesh Phase 4, which cuts the ConfigDb from driver-only
/// nodes). Living in the consolidated LocalDb file is what lets the state replicate to the
/// redundant pair peer, exactly like the alarm store-and-forward buffer — see
/// <see cref="AlarmSfSchema"/>, which this mirrors.
/// </para>
/// <para>
/// Deliberately depends on nothing but <see cref="SqliteConnection"/> so it can be applied
/// to any connection — the host's <c>LocalDbSetup.OnReady</c> in production, and a bare
/// connection in a test — without dragging the DI graph along.
/// </para>
/// <para>
/// <b>The primary key is the alarm identity, and Save is an unconditional upsert onto it.</b>
/// Convergence across the pair is last-writer-wins over the primary key, handled by the
/// LocalDb replication HLC — so the store stacks no application-level last-write-wins on top,
/// the same discipline the alarm-sf sink and the deployment pointer follow.
/// </para>
/// <para>
/// <b>ActiveState has no column</b> — it is re-derived from the live predicate on startup,
/// so persisting it would only risk operators seeing a stale Active on restart. Likewise
/// <c>LastActiveUtc</c> / <c>LastClearedUtc</c> re-derive alongside it and get no columns.
/// <c>LastTransitionUtc</c> is carried by <c>updated_at_utc</c> — the row-write timestamp is
/// the last transition. Timestamps are round-trip ("O") TEXT so lexicographic order is
/// chronological; nullable timestamps and audit fields are stored as NULL.
/// </para>
/// </remarks>
public static class AlarmConditionStateSchema
{
/// <summary>Table holding one condition-state row per scripted-alarm identity.</summary>
public const string StateTable = "alarm_condition_state";
/// <summary>
/// Creates the state table if it does not already exist. Idempotent.
/// </summary>
/// <param name="connection">
/// An already-open connection. <c>ILocalDb.CreateConnection()</c> hands out open,
/// pragma-configured connections — do not call <c>Open()</c> on one.
/// </param>
public static void Apply(SqliteConnection connection)
{
ArgumentNullException.ThrowIfNull(connection);
using var cmd = connection.CreateCommand();
// The four *_state columns are always written to a mapped, non-null string (Enabled/Disabled,
// Acknowledged/Unacknowledged, Confirmed/Unconfirmed, Unshelved/OneShotShelved/TimedShelved),
// so they are NOT NULL. comments_json is never null either — an empty trail is the literal
// "[]". The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm
// has no such history yet. updated_at_utc carries LastTransitionUtc (there is no dedicated
// transition column).
cmd.CommandText = """
CREATE TABLE IF NOT EXISTS alarm_condition_state (
scripted_alarm_id TEXT NOT NULL PRIMARY KEY,
enabled_state TEXT NOT NULL,
acked_state TEXT NOT NULL,
confirmed_state TEXT NOT NULL,
shelving_state TEXT NOT NULL,
shelving_expires_utc TEXT NULL,
last_ack_user TEXT NULL,
last_ack_comment TEXT NULL,
last_ack_utc TEXT NULL,
last_confirm_user TEXT NULL,
last_confirm_comment TEXT NULL,
last_confirm_utc TEXT NULL,
comments_json TEXT NOT NULL,
updated_at_utc TEXT NOT NULL
);
""";
cmd.ExecuteNonQuery();
}
}
src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs
+5 -2
View File
@@ -6,8 +6,9 @@ using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
/// <summary>
/// The <c>onReady</c> callback handed to <c>AddZbLocalDb</c>: creates the deployment-cache and
/// alarm store-and-forward tables and opts them into replication.
/// The <c>onReady</c> callback handed to <c>AddZbLocalDb</c>: creates the deployment-cache,
/// alarm store-and-forward, and scripted-alarm condition-state tables and opts them into
/// replication.
/// </summary>
/// <remarks>
/// <para>
@@ -56,11 +57,13 @@ public static class LocalDbSetup
{
DeploymentCacheSchema.Apply(connection);
AlarmSfSchema.Apply(connection);
AlarmConditionStateSchema.Apply(connection);
}
db.RegisterReplicated(DeploymentCacheSchema.ArtifactsTable);
db.RegisterReplicated(DeploymentCacheSchema.PointerTable);
db.RegisterReplicated(AlarmSfSchema.EventsTable);
db.RegisterReplicated(AlarmConditionStateSchema.StateTable);
// LAST, and only here. This is the one call in OnReady that writes rows.
AlarmSfLegacyMigrator.Migrate(db, configuration);
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs
+264
View File
@@ -0,0 +1,264 @@
using System.Collections.Immutable;
using System.Globalization;
using System.Text.Json;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// <summary>
/// Node-local <see cref="IAlarmStateStore"/> backed by the replicated
/// <see cref="AlarmConditionStateSchema.StateTable"/> table in the node's consolidated LocalDb.
/// Maps the full Part 9 <see cref="AlarmConditionState"/> — Enabled / Acked / Confirmed /
/// Shelving + the ack/confirm audit trail + operator comments.
/// </summary>
/// <remarks>
/// <para>
/// The LocalDb replacement for <c>EfAlarmConditionStateStore</c> (which persisted to the
/// central config DB's <c>ScriptedAlarmState</c> table). Per-cluster mesh Phase 4 cuts the
/// ConfigDb from driver-only nodes, so this state moves into the same replicated LocalDb
/// file the Phase-2 alarm store-and-forward buffer already lives in — a node's condition
/// state then mirrors to its redundant pair peer instead of depending on central SQL.
/// </para>
/// <para>
/// <b>The round-trip is byte-identical to the EF store</b> — the enum↔string mappers and the
/// comment serialization below are duplicated verbatim from it (it is deleted in a later
/// task; the ported round-trip tests prove parity). The persistence rules carry over
/// unchanged:
/// </para>
/// <para>
/// <b>ActiveState is NOT persisted</b> — there is no column; on <see cref="LoadAsync"/> it is
/// restored as <see cref="AlarmActiveState.Inactive"/> and the engine re-derives it from the
/// live predicate. <b>LastTransitionUtc ↔ updated_at_utc</b>: no dedicated transition column,
/// so the last transition rides the row-write timestamp. <b>LastActiveUtc / LastClearedUtc</b>
/// have no columns and default to <c>null</c> on load. <see cref="AlarmConditionState.Comments"/>
/// serializes to/from <c>comments_json</c>; an empty list round-trips as <c>"[]"</c>.
/// </para>
/// <para>
/// <b>Save is a single unconditional PK upsert.</b> No application-level last-write-wins or
/// HLC comparison is layered on top — the LocalDb replication HLC handles convergence, and an
/// unconditional upsert keyed by the primary key is HLC-safe (the same discipline the
/// alarm-sf sink and the deployment pointer follow).
/// </para>
/// </remarks>
public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore
{
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
// Fixed column order shared by every read — the reader map below indexes by these ordinals.
private const string SelectColumns =
"scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, " +
"shelving_expires_utc, last_ack_user, last_ack_comment, last_ack_utc, " +
"last_confirm_user, last_confirm_comment, last_confirm_utc, comments_json, updated_at_utc";
private readonly ILocalDb _db;
private readonly ILogger<LocalDbAlarmConditionStateStore> _logger;
/// <summary>Initializes a new instance of the <see cref="LocalDbAlarmConditionStateStore"/> class.</summary>
/// <param name="db">
/// The node's local database. Its <see cref="AlarmConditionStateSchema.StateTable"/> table
/// must already exist and be registered for replication — the host does both in
/// <c>LocalDbSetup.OnReady</c>, before any consumer can resolve this store.
/// </param>
/// <param name="logger">The logger instance.</param>
public LocalDbAlarmConditionStateStore(ILocalDb db, ILogger<LocalDbAlarmConditionStateStore> logger)
{
_db = db ?? throw new ArgumentNullException(nameof(db));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
/// <inheritdoc />
public async Task<AlarmConditionState?> LoadAsync(string alarmId, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(alarmId);
var rows = await _db.QueryAsync(
$"SELECT {SelectColumns} FROM alarm_condition_state WHERE scripted_alarm_id = @Id",
MapRow,
new { Id = alarmId },
ct).ConfigureAwait(false);
return rows.Count > 0 ? rows[0] : null;
}
/// <inheritdoc />
public async Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct)
{
var rows = await _db.QueryAsync(
$"SELECT {SelectColumns} FROM alarm_condition_state",
MapRow,
parameters: null,
ct).ConfigureAwait(false);
return rows;
}
/// <inheritdoc />
public async Task SaveAsync(AlarmConditionState state, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(state);
// A single unconditional upsert keyed on the alarm identity. Convergence across the pair is
// last-writer-wins over the primary key, handled by the LocalDb replication HLC — no
// app-level LWW here (see the class remarks).
await _db.ExecuteAsync(
"""
INSERT INTO alarm_condition_state
(scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state,
shelving_expires_utc, last_ack_user, last_ack_comment, last_ack_utc,
last_confirm_user, last_confirm_comment, last_confirm_utc, comments_json, updated_at_utc)
VALUES
(@Id, @Enabled, @Acked, @Confirmed, @Shelving,
@ShelvingExpires, @LastAckUser, @LastAckComment, @LastAckUtc,
@LastConfirmUser, @LastConfirmComment, @LastConfirmUtc, @CommentsJson, @UpdatedAtUtc)
ON CONFLICT(scripted_alarm_id) DO UPDATE SET
enabled_state = excluded.enabled_state,
acked_state = excluded.acked_state,
confirmed_state = excluded.confirmed_state,
shelving_state = excluded.shelving_state,
shelving_expires_utc = excluded.shelving_expires_utc,
last_ack_user = excluded.last_ack_user,
last_ack_comment = excluded.last_ack_comment,
last_ack_utc = excluded.last_ack_utc,
last_confirm_user = excluded.last_confirm_user,
last_confirm_comment = excluded.last_confirm_comment,
last_confirm_utc = excluded.last_confirm_utc,
comments_json = excluded.comments_json,
updated_at_utc = excluded.updated_at_utc
""",
new
{
Id = state.AlarmId,
Enabled = MapEnabledToColumn(state.Enabled),
Acked = MapAckedToColumn(state.Acked),
Confirmed = MapConfirmedToColumn(state.Confirmed),
Shelving = MapShelvingToColumn(state.Shelving.Kind),
ShelvingExpires = FormatNullable(state.Shelving.UnshelveAtUtc),
LastAckUser = state.LastAckUser,
LastAckComment = state.LastAckComment,
LastAckUtc = FormatNullable(state.LastAckUtc),
LastConfirmUser = state.LastConfirmUser,
LastConfirmComment = state.LastConfirmComment,
LastConfirmUtc = FormatNullable(state.LastConfirmUtc),
CommentsJson = SerializeComments(state.Comments),
// No dedicated transition column — persist LastTransitionUtc into updated_at_utc.
UpdatedAtUtc = Format(state.LastTransitionUtc),
},
ct).ConfigureAwait(false);
_logger.LogTrace("Persisted alarm-condition state for {AlarmId}", state.AlarmId);
}
/// <inheritdoc />
public async Task RemoveAsync(string alarmId, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(alarmId);
await _db.ExecuteAsync(
"DELETE FROM alarm_condition_state WHERE scripted_alarm_id = @Id",
new { Id = alarmId },
ct).ConfigureAwait(false);
}
private static AlarmConditionState MapRow(SqliteDataReader r) => new(
AlarmId: r.GetString(0),
Enabled: string.Equals(r.GetString(1), "Disabled", StringComparison.Ordinal)
? AlarmEnabledState.Disabled
: AlarmEnabledState.Enabled, // unknown string → Enabled (safe default)
// Active is not persisted — the engine re-derives it from the predicate at startup.
Active: AlarmActiveState.Inactive,
Acked: string.Equals(r.GetString(2), "Acknowledged", StringComparison.Ordinal)
? AlarmAckedState.Acknowledged
: AlarmAckedState.Unacknowledged, // unknown string → Unacknowledged (safe default)
Confirmed: string.Equals(r.GetString(3), "Confirmed", StringComparison.Ordinal)
? AlarmConfirmedState.Confirmed
: AlarmConfirmedState.Unconfirmed, // unknown string → Unconfirmed (safe default)
Shelving: new ShelvingState(MapShelvingFromColumn(r.GetString(4)), ParseNullable(r, 5)),
// No transition column — updated_at_utc carries the last transition timestamp.
LastTransitionUtc: Parse(r.GetString(13)),
// LastActiveUtc / LastClearedUtc have no columns — they re-derive with Active, so null on load.
LastActiveUtc: null,
LastClearedUtc: null,
LastAckUtc: ParseNullable(r, 8),
LastAckUser: r.IsDBNull(6) ? null : r.GetString(6),
LastAckComment: r.IsDBNull(7) ? null : r.GetString(7),
LastConfirmUtc: ParseNullable(r, 11),
LastConfirmUser: r.IsDBNull(9) ? null : r.GetString(9),
LastConfirmComment: r.IsDBNull(10) ? null : r.GetString(10),
Comments: DeserializeComments(r.GetString(12)));
// Timestamps are stored as round-trip ("O") TEXT so lexicographic order stays chronological and
// the DateTimeKind.Utc round-trips exactly.
private static string Format(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture);
private static string? FormatNullable(DateTime? value) => value?.ToString("O", CultureInfo.InvariantCulture);
private static DateTime Parse(string value) =>
DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind);
private static DateTime? ParseNullable(SqliteDataReader r, int ordinal) =>
r.IsDBNull(ordinal) ? null : Parse(r.GetString(ordinal));
private static string MapEnabledToColumn(AlarmEnabledState enabled)
=> enabled == AlarmEnabledState.Enabled ? "Enabled" : "Disabled";
private static string MapAckedToColumn(AlarmAckedState acked)
=> acked == AlarmAckedState.Acknowledged ? "Acknowledged" : "Unacknowledged";
private static string MapConfirmedToColumn(AlarmConfirmedState confirmed)
=> confirmed == AlarmConfirmedState.Confirmed ? "Confirmed" : "Unconfirmed";
private static string MapShelvingToColumn(ShelvingKind kind) => kind switch
{
ShelvingKind.OneShot => "OneShotShelved",
ShelvingKind.Timed => "TimedShelved",
_ => "Unshelved",
};
private static ShelvingKind MapShelvingFromColumn(string column) => column switch
{
"OneShotShelved" => ShelvingKind.OneShot,
"TimedShelved" => ShelvingKind.Timed,
_ => ShelvingKind.Unshelved, // unknown string → Unshelved (safe default)
};
private static string SerializeComments(ImmutableList<AlarmComment> comments)
{
if (comments.IsEmpty) return "[]";
var dtos = comments.Select(c => new CommentDto
{
// AlarmComment.TimestampUtc must be DateTimeKind.Utc for correct ISO-8601 round-trip;
// the engine always creates AlarmComment instances with Utc kind.
TimestampUtc = c.TimestampUtc,
User = c.User,
Kind = c.Kind,
Text = c.Text,
});
return JsonSerializer.Serialize(dtos, JsonOptions);
}
private static ImmutableList<AlarmComment> DeserializeComments(string? json)
{
if (string.IsNullOrWhiteSpace(json)) return ImmutableList<AlarmComment>.Empty;
var dtos = JsonSerializer.Deserialize<List<CommentDto>>(json, JsonOptions);
if (dtos is null || dtos.Count == 0) return ImmutableList<AlarmComment>.Empty;
return dtos
.Select(d => new AlarmComment(d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty))
.ToImmutableList();
}
/// <summary>Stable on-disk shape for a persisted <see cref="AlarmComment"/> in <c>comments_json</c>.</summary>
private sealed class CommentDto
{
/// <summary>When the comment was recorded (UTC).</summary>
public DateTime TimestampUtc { get; set; }
/// <summary>Identity of the actor that wrote the comment.</summary>
public string? User { get; set; }
/// <summary>Human-readable classification of the comment (Acknowledge, Confirm, …).</summary>
public string? Kind { get; set; }
/// <summary>Operator-supplied or engine-generated comment text.</summary>
public string? Text { get; set; }
}
}