diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs index 38b3acf5..1ec6b043 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs @@ -1,11 +1,13 @@ +using System.Globalization; +using System.Text.Json; using Microsoft.Data.Sqlite; namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// -/// 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. +/// DDL for the Part 9 scripted-alarm condition state: one fixed-size row per alarm identity +/// holding the operator-supplied state (Enabled / Acked / Confirmed / Shelving) plus the +/// ack/confirm audit fields, and a separate append-only table holding each alarm's comment trail. /// /// /// @@ -34,14 +36,55 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// the last transition. Timestamps are round-trip ("O") TEXT so lexicographic order is /// chronological; nullable timestamps and audit fields are stored as NULL. /// +/// +/// The comment trail is NOT on the condition row. It used to be serialized whole into +/// comments_json, so every transition rewrote — and replicated — the alarm's entire +/// history: O(n) per transition, O(n²) over the alarm's life (review R05:4). It now lives in +/// , one row per entry keyed by (scripted_alarm_id, seq), +/// where seq is the entry's ordinal in the alarm's trail. A transition inserts only the +/// entries it appended, so the condition row stays a fixed size and a transition's write is +/// O(1) however long the history is. comments_json is kept, always written as +/// "[]", so the condition table's shape — and with it the pair's replication digest +/// for that table — does not change and an old-shape row stays readable; +/// moves any inline trail across. +/// +/// +/// Why the ordinal, and not a content hash, is the key. Both halves of a redundant pair +/// run the engine and both apply every operator command, each stamping the entry with its own +/// clock. A content-derived id would keep TWO rows for every logical entry once the pair +/// converged. Keyed by ordinal, the two nodes' copies of entry n share one primary key +/// and LWW collapses them to one row — the convergence the whole-row LWW gave before, now per +/// entry. Entries go in with INSERT OR IGNORE (), so an +/// entry already held — written locally, or arrived from the peer — is never rewritten. +/// /// public static class AlarmConditionStateSchema { /// Table holding one condition-state row per scripted-alarm identity. public const string StateTable = "alarm_condition_state"; + /// Append-only table holding one row per comment-trail entry, keyed by alarm id + ordinal. + public const string CommentsTable = "alarm_condition_comment"; + + /// What comments_json holds once the trail lives in . + public const string EmptyInlineComments = "[]"; + /// - /// Creates the state table if it does not already exist. Idempotent. + /// The one append statement for , shared by the migration and the + /// store. OR IGNORE is what makes the trail append-only: an ordinal already held is + /// never rewritten. Parameters: @Id @Seq @Ts @User @Kind @Text. + /// + public const string InsertCommentSql = + """ + INSERT OR IGNORE INTO alarm_condition_comment + (scripted_alarm_id, seq, timestamp_utc, user_name, kind, text) + VALUES (@Id, @Seq, @Ts, @User, @Kind, @Text) + """; + + private static readonly JsonSerializerOptions InlineJsonOptions = new(JsonSerializerDefaults.Web); + + /// + /// Creates the state table and the comment table if they do not already exist. Idempotent. /// /// /// An already-open connection. ILocalDb.CreateConnection() hands out open, @@ -55,10 +98,11 @@ public static class AlarmConditionStateSchema // 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). + // so they are NOT NULL. comments_json is never null either — it is the literal "[]" since the + // trail moved to alarm_condition_comment (an old-shape row may still carry an inline trail + // until MigrateInlineComments or the next save moves it). 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, @@ -76,7 +120,151 @@ public static class AlarmConditionStateSchema comments_json TEXT NOT NULL, updated_at_utc TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS alarm_condition_comment ( + scripted_alarm_id TEXT NOT NULL, + seq INTEGER NOT NULL, + timestamp_utc TEXT NOT NULL, + user_name TEXT NOT NULL, + kind TEXT NOT NULL, + text TEXT NOT NULL, + PRIMARY KEY (scripted_alarm_id, seq) + ); """; cmd.ExecuteNonQuery(); } + + /// + /// Moves every condition row's inline comments_json trail into + /// and resets the column to "[]". Idempotent. + /// + /// + /// + /// Run it AFTER the tables are registered for replication — it writes rows, and a + /// row written before the capture triggers exist never reaches the peer (the same + /// DDL → register → writes rule AlarmSfLegacyMigrator follows). + /// + /// + /// Each entry lands at its ordinal in the inline array via , + /// so a re-run — or a row whose trail was already (partly) moved — neither duplicates nor + /// reorders anything, and a second run finds no non-empty inline column and writes + /// nothing. Each row moves in its own transaction together with its column reset, so a + /// crash leaves every row either fully moved or untouched. A row whose inline JSON does + /// not parse is left exactly as it is (the pre-split store could not load it either) + /// rather than failing host start. + /// + /// + /// An already-open connection from ILocalDb.CreateConnection(). + /// The number of condition rows whose inline trail was moved. + public static int MigrateInlineComments(SqliteConnection connection) + { + ArgumentNullException.ThrowIfNull(connection); + + var pending = new List<(string AlarmId, string Json)>(); + using (var select = connection.CreateCommand()) + { + select.CommandText = + "SELECT scripted_alarm_id, comments_json FROM alarm_condition_state WHERE comments_json <> @Empty"; + select.Parameters.AddWithValue("@Empty", EmptyInlineComments); + using var reader = select.ExecuteReader(); + while (reader.Read()) + pending.Add((reader.GetString(0), reader.GetString(1))); + } + + var migrated = 0; + foreach (var (alarmId, json) in pending) + { + IReadOnlyList entries; + try + { + entries = ParseInlineComments(json); + } + catch (JsonException) + { + continue; + } + + using var tx = connection.BeginTransaction(); + for (var seq = 0; seq < entries.Count; seq++) + { + using var insert = connection.CreateCommand(); + insert.Transaction = tx; + insert.CommandText = InsertCommentSql; + insert.Parameters.AddWithValue("@Id", alarmId); + insert.Parameters.AddWithValue("@Seq", seq); + insert.Parameters.AddWithValue("@Ts", FormatTimestamp(entries[seq].TimestampUtc)); + insert.Parameters.AddWithValue("@User", entries[seq].User); + insert.Parameters.AddWithValue("@Kind", entries[seq].Kind); + insert.Parameters.AddWithValue("@Text", entries[seq].Text); + insert.ExecuteNonQuery(); + } + + using (var clear = connection.CreateCommand()) + { + clear.Transaction = tx; + clear.CommandText = + "UPDATE alarm_condition_state SET comments_json = @Empty WHERE scripted_alarm_id = @Id"; + clear.Parameters.AddWithValue("@Empty", EmptyInlineComments); + clear.Parameters.AddWithValue("@Id", alarmId); + clear.ExecuteNonQuery(); + } + + tx.Commit(); + migrated++; + } + + return migrated; + } + + /// Formats a comment timestamp for (round-trip "O" text). + /// The timestamp. + /// The round-trip text form. + public static string FormatTimestamp(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture); + + /// Parses a timestamp, preserving its . + /// The round-trip text form. + /// The timestamp. + public static DateTime ParseTimestamp(string value) => + DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + + /// + /// Parses a legacy inline comments_json trail exactly as the pre-split store did + /// (; null strings become empty). + /// + /// The column value. + /// The entries in trail order; empty for a blank or "[]" column. + /// The column is not a valid trail. + public static IReadOnlyList ParseInlineComments(string? json) + { + if (string.IsNullOrWhiteSpace(json)) return []; + var dtos = JsonSerializer.Deserialize>(json, InlineJsonOptions); + if (dtos is null || dtos.Count == 0) return []; + return dtos + .Select(d => new InlineComment( + d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty)) + .ToList(); + } + + /// One comment-trail entry as stored — the persistence-side mirror of the engine's AlarmComment. + /// When the action happened. + /// Identity of the actor. + /// Classification (Acknowledge, Confirm, AddComment, …). + /// Operator-supplied or engine-generated text. + public sealed record InlineComment(DateTime TimestampUtc, string User, string Kind, string Text); + + /// The pre-split on-disk shape of one entry in comments_json. + private sealed class InlineCommentDto + { + /// When the comment was recorded. + public DateTime TimestampUtc { get; set; } + + /// Identity of the actor. + public string? User { get; set; } + + /// Classification of the entry. + public string? Kind { get; set; } + + /// Entry text. + public string? Text { get; set; } + } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs index 3a4129a2..c841db21 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbSetup.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; /// /// The onReady callback handed to AddZbLocalDb: creates the deployment-cache, -/// alarm store-and-forward, and scripted-alarm condition-state tables and — when this node +/// alarm store-and-forward, and scripted-alarm condition-state + comment tables and — when this node /// actually replicates — opts them into change capture. /// /// @@ -21,7 +21,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; public static class LocalDbSetup { /// - /// The four tables that participate in replication — registered when this node is configured + /// The five tables that participate in replication — registered when this node is configured /// to replicate, and deregistered (stale capture triggers dropped) when it is not. /// /// @@ -36,6 +36,7 @@ public static class LocalDbSetup DeploymentCacheSchema.PointerTable, AlarmSfSchema.EventsTable, AlarmConditionStateSchema.StateTable, + AlarmConditionStateSchema.CommentsTable, ]; /// @@ -47,8 +48,9 @@ public static class LocalDbSetup /// RegisterReplicated is what installs the three AFTER triggers that capture /// changes into the oplog. Any row written before that call is never captured, so it /// never reaches the peer — silently, and permanently, because nothing ever revisits - /// history. therefore runs last, after every - /// registration — it writes rows, and they must be captured like any other write. + /// history. and + /// therefore run last, after every + /// registration — they write rows, and those must be captured like any other write. /// /// /// Registration is CONDITIONAL on . It used @@ -68,7 +70,7 @@ public static class LocalDbSetup /// /// /// A failure part-way through either loop fails host start, and that is deliberate. - /// Neither branch catches, so a throw on the third of four tables leaves two flipped and + /// Neither branch catches, so a throw on the third of five tables leaves two flipped and /// propagates out of the AddZbLocalDb singleton factory — a node that came up /// anyway would be capturing (or not) on an arbitrary subset, which is silently partial /// replication with nothing in the log to say so. Both library calls are idempotent, so @@ -116,7 +118,15 @@ public static class LocalDbSetup else RemoveStaleCapture(db, logger ?? Serilog.Log.Logger); - // LAST, and only here. This is the one call in OnReady that writes rows. + // LAST, and only here. These are the only calls in OnReady that write rows. + // Moves any pre-split inline comment trail (alarm_condition_state.comments_json) into the + // append-only alarm_condition_comment table (review R05:4). Idempotent: a boot after the + // first finds no inline trail and writes nothing. + using (var connection = db.CreateConnection()) + { + AlarmConditionStateSchema.MigrateInlineComments(connection); + } + AlarmSfLegacyMigrator.Migrate(db, configuration); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs index d540f52c..8437574d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/LocalDbAlarmConditionStateStore.cs @@ -1,6 +1,4 @@ using System.Collections.Immutable; -using System.Globalization; -using System.Text.Json; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Logging; using ZB.MOM.WW.LocalDb; @@ -11,9 +9,11 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; /// /// Node-local backed by the replicated -/// table in the node's consolidated LocalDb. -/// Maps the full Part 9 — Enabled / Acked / Confirmed / -/// Shelving + the ack/confirm audit trail + operator comments. +/// and +/// tables in the node's consolidated +/// LocalDb. Maps the full Part 9 — Enabled / Acked / Confirmed / +/// Shelving + the ack/confirm audit fields on the condition row, and the append-only comment +/// trail one row per entry. /// /// /// @@ -24,44 +24,52 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; /// state then mirrors to its redundant pair peer instead of depending on central SQL. /// /// -/// The round-trip is byte-identical to the EF store — 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: -/// -/// /// ActiveState is NOT persisted — there is no column; on it is /// restored as and the engine re-derives it from the /// live predicate. LastTransitionUtc ↔ updated_at_utc: no dedicated transition column, /// so the last transition rides the row-write timestamp. LastActiveUtc / LastClearedUtc -/// have no columns and default to null on load. -/// serializes to/from comments_json; an empty list round-trips as "[]". +/// have no columns and default to null on load. /// /// -/// Save is a single unconditional PK upsert. 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). +/// The comment trail is written incrementally (review R05:4). Each save appends only +/// the entries past the highest ordinal already stored for the alarm, so a transition's write +/// — and its replication payload — is constant however long the history is; the condition +/// row's comments_json is always written as "[]". The "already stored" boundary +/// is read from the table inside the save's transaction rather than cached, so the store +/// holds no per-alarm state and an entry the pair peer already wrote at an ordinal is kept, +/// not duplicated (see for why the ordinal is the key). +/// +/// +/// Old-shape rows load exactly as before. A row still carrying an inline +/// comments_json trail (written by the pre-split store and not yet moved by +/// ) is read through the same +/// parser the old store used, merged by ordinal with whatever the comment table holds (the +/// table wins an ordinal both hold), and moved to the table on that alarm's next save. +/// +/// +/// Save is one transaction: the comment append and an unconditional PK upsert of the +/// condition row. No application-level last-write-wins or HLC comparison is layered on top — +/// the LocalDb replication HLC handles convergence. /// /// 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 const string SelectCommentColumns = "scripted_alarm_id, seq, timestamp_utc, user_name, kind, text"; + private readonly ILocalDb _db; private readonly ILogger _logger; /// Initializes a new instance of the class. /// - /// The node's local database. Its table - /// must already exist and be registered for replication — the host does both in - /// LocalDbSetup.OnReady, before any consumer can resolve this store. + /// The node's local database. Its condition-state and comment tables must already exist and + /// be registered for replication — the host does both in LocalDbSetup.OnReady, before + /// any consumer can resolve this store. /// /// The logger instance. public LocalDbAlarmConditionStateStore(ILocalDb db, ILogger logger) @@ -79,7 +87,15 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore MapRow, new { Id = alarmId }, ct).ConfigureAwait(false); - return rows.Count > 0 ? rows[0] : null; + if (rows.Count == 0) return null; + + var comments = await _db.QueryAsync( + $"SELECT {SelectCommentColumns} FROM alarm_condition_comment WHERE scripted_alarm_id = @Id ORDER BY seq", + MapComment, + new { Id = alarmId }, + ct).ConfigureAwait(false); + + return WithComments(rows[0], comments); } /// @@ -90,7 +106,19 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore MapRow, parameters: null, ct).ConfigureAwait(false); - return rows; + + var comments = await _db.QueryAsync( + $"SELECT {SelectCommentColumns} FROM alarm_condition_comment ORDER BY scripted_alarm_id, seq", + MapComment, + parameters: null, + ct).ConfigureAwait(false); + var byAlarm = comments + .GroupBy(c => c.AlarmId, StringComparer.Ordinal) + .ToDictionary(g => g.Key, g => (IReadOnlyList)g.ToList(), StringComparer.Ordinal); + + return rows + .Select(r => WithComments(r, byAlarm.TryGetValue(r.State.AlarmId, out var c) ? c : [])) + .ToList(); } /// @@ -98,10 +126,40 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore { ArgumentNullException.ThrowIfNull(state); + await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false); + + // Append only what lies past the highest ordinal already stored. An indexed MAX over the + // (scripted_alarm_id, seq) primary key — O(log n), and a read, so it costs no replication. + var maxSeq = (await tx.QueryAsync( + "SELECT COALESCE(MAX(seq), -1) FROM alarm_condition_comment WHERE scripted_alarm_id = @Id", + static r => r.GetInt64(0), + new { Id = state.AlarmId }, + ct).ConfigureAwait(false))[0]; + + for (var seq = (int)Math.Max(maxSeq + 1, 0); seq < state.Comments.Count; seq++) + { + var c = state.Comments[seq]; + await tx.ExecuteAsync( + AlarmConditionStateSchema.InsertCommentSql, + new + { + Id = state.AlarmId, + Seq = seq, + // AlarmComment.TimestampUtc must be DateTimeKind.Utc for a Utc round-trip; the + // engine always creates comments with Utc kind. + Ts = AlarmConditionStateSchema.FormatTimestamp(c.TimestampUtc), + User = c.User, + Kind = c.Kind, + Text = c.Text, + }, + ct).ConfigureAwait(false); + } + // 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( + // app-level LWW here (see the class remarks). comments_json is always "[]": the trail lives + // in alarm_condition_comment, which is what keeps this row a fixed size. + await tx.ExecuteAsync( """ INSERT INTO alarm_condition_state (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, @@ -140,12 +198,14 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore LastConfirmUser = state.LastConfirmUser, LastConfirmComment = state.LastConfirmComment, LastConfirmUtc = FormatNullable(state.LastConfirmUtc), - CommentsJson = SerializeComments(state.Comments), + CommentsJson = AlarmConditionStateSchema.EmptyInlineComments, // No dedicated transition column — persist LastTransitionUtc into updated_at_utc. UpdatedAtUtc = Format(state.LastTransitionUtc), }, ct).ConfigureAwait(false); + await tx.CommitAsync(ct).ConfigureAwait(false); + _logger.LogTrace("Persisted alarm-condition state for {AlarmId}", state.AlarmId); } @@ -153,47 +213,93 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore public async Task RemoveAsync(string alarmId, CancellationToken ct) { ArgumentNullException.ThrowIfNull(alarmId); - await _db.ExecuteAsync( + + // The trail goes with the condition, as it did when it was a column of the row. + await using var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false); + await tx.ExecuteAsync( + "DELETE FROM alarm_condition_comment WHERE scripted_alarm_id = @Id", + new { Id = alarmId }, + ct).ConfigureAwait(false); + await tx.ExecuteAsync( "DELETE FROM alarm_condition_state WHERE scripted_alarm_id = @Id", new { Id = alarmId }, ct).ConfigureAwait(false); + await tx.CommitAsync(ct).ConfigureAwait(false); } - private static AlarmConditionState MapRow(SqliteDataReader r) => new( + /// A condition row as read, before its trail is attached. + private sealed record ConditionRow(AlarmConditionState State, string InlineCommentsJson); + + /// One alarm_condition_comment row. + private sealed record StoredComment(string AlarmId, long Seq, AlarmComment Comment); + + /// + /// Attaches the trail: the comment table's entries by ordinal, plus — for an old-shape row not + /// yet migrated — any inline entry at an ordinal the table does not hold. + /// + private static AlarmConditionState WithComments(ConditionRow row, IReadOnlyList stored) + { + var inline = AlarmConditionStateSchema.ParseInlineComments(row.InlineCommentsJson); + if (inline.Count == 0) + return row.State with { Comments = stored.Select(s => s.Comment).ToImmutableList() }; + + var merged = new SortedDictionary(); + foreach (var s in stored) + merged[s.Seq] = s.Comment; + for (var i = 0; i < inline.Count; i++) + { + var c = inline[i]; + merged.TryAdd(i, new AlarmComment(c.TimestampUtc, c.User, c.Kind, c.Text)); + } + + return row.State with { Comments = merged.Values.ToImmutableList() }; + } + + private static StoredComment MapComment(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))); + Seq: r.GetInt64(1), + Comment: new AlarmComment( + AlarmConditionStateSchema.ParseTimestamp(r.GetString(2)), + r.GetString(3), + r.GetString(4), + r.GetString(5))); + + private static ConditionRow MapRow(SqliteDataReader r) => new( + new AlarmConditionState( + 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: ImmutableList.Empty), + InlineCommentsJson: 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 Format(DateTime value) => AlarmConditionStateSchema.FormatTimestamp(value); - private static string? FormatNullable(DateTime? value) => value?.ToString("O", CultureInfo.InvariantCulture); + private static string? FormatNullable(DateTime? value) => value is { } v ? Format(v) : null; - private static DateTime Parse(string value) => - DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + private static DateTime Parse(string value) => AlarmConditionStateSchema.ParseTimestamp(value); private static DateTime? ParseNullable(SqliteDataReader r, int ordinal) => r.IsDBNull(ordinal) ? null : Parse(r.GetString(ordinal)); @@ -220,45 +326,4 @@ public sealed class LocalDbAlarmConditionStateStore : IAlarmStateStore "TimedShelved" => ShelvingKind.Timed, _ => ShelvingKind.Unshelved, // unknown string → Unshelved (safe default) }; - - private static string SerializeComments(ImmutableList 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 DeserializeComments(string? json) - { - if (string.IsNullOrWhiteSpace(json)) return ImmutableList.Empty; - var dtos = JsonSerializer.Deserialize>(json, JsonOptions); - if (dtos is null || dtos.Count == 0) return ImmutableList.Empty; - return dtos - .Select(d => new AlarmComment(d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty)) - .ToImmutableList(); - } - - /// Stable on-disk shape for a persisted in comments_json. - private sealed class CommentDto - { - /// When the comment was recorded (UTC). - public DateTime TimestampUtc { get; set; } - - /// Identity of the actor that wrote the comment. - public string? User { get; set; } - - /// Human-readable classification of the comment (Acknowledge, Confirm, …). - public string? Kind { get; set; } - - /// Operator-supplied or engine-generated comment text. - public string? Text { get; set; } - } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs index 028bc49f..8e9cf432 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmConditionStateSchemaTests.cs @@ -34,6 +34,24 @@ public sealed class AlarmConditionStateSchemaTests ((long)cmd.ExecuteScalar()!).ShouldBe(1); } + /// Apply also creates the append-only comment table, keyed by alarm id + ordinal. + [Fact] + public void Apply_creates_the_comment_table_with_a_composite_key() + { + using var connection = OpenInMemory(); + + AlarmConditionStateSchema.Apply(connection); + + using var cmd = connection.CreateCommand(); + cmd.CommandText = "SELECT name FROM pragma_table_info(@Name) WHERE pk > 0 ORDER BY pk"; + cmd.Parameters.AddWithValue("@Name", AlarmConditionStateSchema.CommentsTable); + var pk = new List(); + using (var reader = cmd.ExecuteReader()) + while (reader.Read()) pk.Add(reader.GetString(0)); + + pk.ShouldBe(["scripted_alarm_id", "seq"]); + } + /// Applying twice on the same connection does not throw (CREATE TABLE IF NOT EXISTS). [Fact] public void Apply_is_idempotent() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs new file mode 100644 index 00000000..26b1fe14 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmConditionCommentConvergenceTests.cs @@ -0,0 +1,117 @@ +using System.Collections.Immutable; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb; + +/// +/// Review R05:4 — the scripted-alarm comment trail moved off the condition row into its own +/// append-only alarm_condition_comment table. This proves the move kept the property that +/// mattered: the trail still reaches the redundant pair peer, over a real loopback h2c transport +/// through the real fail-closed interceptor, with both nodes initialised by the production +/// LocalDbSetup.OnReady. +/// +/// +/// State is written and read through the production +/// on each node, so the ordinal keying, the incremental append and the load-side merge under +/// test are the ones production runs. +/// +[Collection("LocalDbPairConvergence")] +public sealed class AlarmConditionCommentConvergenceTests +{ + private static readonly DateTime T0 = new(2026, 06, 10, 12, 00, 00, DateTimeKind.Utc); + + private static LocalDbAlarmConditionStateStore Store(ILocalDb db) => + new(db, NullLogger.Instance); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static async Task CommentCountAsync(ILocalDb db, string alarmId) => + (await Store(db).LoadAsync(alarmId, Ct))?.Comments.Count ?? -1; + + [Fact] + public async Task CommentsWrittenOnA_ArriveOnB_AndBsAppendArrivesBackOnA() + { + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + var storeA = Store(pair.A); + var state = AlarmConditionState.Fresh("eq/alarm-1", T0) with + { + Active = AlarmActiveState.Active, + Acked = AlarmAckedState.Unacknowledged, + }; + state = Part9StateMachine.ApplyAddComment(state, "jane", "first look", T0.AddMinutes(1)).State; + state = Part9StateMachine.ApplyAcknowledge(state, "bob", "on it", T0.AddMinutes(2)).State; + await storeA.SaveAsync(state, Ct); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.B, "eq/alarm-1") == 2, + "node B to hold node A's two comment-trail entries"); + + var onB = await Store(pair.B).LoadAsync("eq/alarm-1", Ct); + onB.ShouldNotBeNull(); + onB.Comments.ShouldBe(state.Comments); + onB.Acked.ShouldBe(AlarmAckedState.Acknowledged); + + // The reverse direction, as an incremental append: B adds one entry to the trail it loaded, + // and only that entry has to travel for A to see the full trail. + var appended = Part9StateMachine.ApplyAddComment(onB, "amy", "from B", T0.AddMinutes(3)).State; + await Store(pair.B).SaveAsync(appended, Ct); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.A, "eq/alarm-1") == 3, + "node A to receive the entry node B appended"); + + (await storeA.LoadAsync("eq/alarm-1", Ct))!.Comments.ShouldBe(appended.Comments); + } + + [Fact] + public async Task BothNodesRecordingTheSameEntry_ConvergeToOneEntry_NotTwo() + { + // Both pair nodes run the engine and both apply every operator command, each stamping the + // entry with its own clock. Ordinal keying is what collapses the two copies to one entry; + // a content-derived key would leave the trail doubled on both nodes. + await using var pair = new LocalDbPairHarness(); + await pair.StartAsync(); + + var fresh = AlarmConditionState.Fresh("eq/alarm-2", T0); + var onA = Part9StateMachine.ApplyAddComment(fresh, "jane", "checked", T0.AddSeconds(1)).State; + var onB = Part9StateMachine.ApplyAddComment(fresh, "jane", "checked", T0.AddSeconds(1).AddMilliseconds(7)).State; + await Store(pair.A).SaveAsync(onA, Ct); + await Store(pair.B).SaveAsync(onB, Ct); + + // Positive control: a DISTINCT second alarm written only on B. Without it "one entry on + // both" would also hold for two databases that never replicated at all. + var onlyOnB = Part9StateMachine.ApplyAddComment( + AlarmConditionState.Fresh("eq/only-on-b", T0), "amy", "b only", T0.AddSeconds(2)).State; + await Store(pair.B).SaveAsync(onlyOnB, Ct); + + await LocalDbPairHarness.WaitUntilAsync( + async () => await CommentCountAsync(pair.A, "eq/only-on-b") == 1, + "node A to receive node B's distinct alarm"); + + // LWW picks ONE node's copy and both nodes end up holding that same copy. + await LocalDbPairHarness.WaitUntilAsync( + async () => + { + var a = await Store(pair.A).LoadAsync("eq/alarm-2", Ct); + var b = await Store(pair.B).LoadAsync("eq/alarm-2", Ct); + return a is not null && b is not null + && a.Comments.Count == 1 && b.Comments.Count == 1 + && a.Comments[0] == b.Comments[0]; + }, + "both nodes to agree on a single copy of the shared entry"); + + // Held, not merely reached: nothing late may double the trail on either node. + (await LocalDbPairHarness.HeldWithinAsync( + async () => await CommentCountAsync(pair.A, "eq/alarm-2") != 1 + || await CommentCountAsync(pair.B, "eq/alarm-2") != 1, + TimeSpan.FromSeconds(2))) + .ShouldBeFalse("the shared entry must stay one entry on both nodes, never two"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs index 197439df..280ac35a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbSetupTests.cs @@ -134,7 +134,7 @@ public sealed class LocalDbSetupTests : IDisposable TestContext.Current.CancellationToken); [Fact] - public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFourReplicatedTables() + public void OnReady_WhenReplicationIsConfigured_RegistersExactlyTheFiveReplicatedTables() { // Both directions are load-bearing. "No fewer" catches a dropped RegisterReplicated call // (that table then never replicates). "No more" catches an accidental registration — @@ -142,11 +142,11 @@ public sealed class LocalDbSetupTests : IDisposable var db = Boot(InitiatorReplication); db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact] - public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFourTables() + public void OnReady_WhenOnlyTheSyncApiKeyIsSet_StillRegistersTheFiveTables() { // The passive half of a pair carries the key and nothing else. If the predicate were // "PeerAddress is set", this node would install no capture triggers and its OWN writes @@ -155,7 +155,7 @@ public sealed class LocalDbSetupTests : IDisposable var db = Boot(PassiveReplication); db.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact] @@ -216,6 +216,53 @@ public sealed class LocalDbSetupTests : IDisposable db.ReplicatedTables["alarm_condition_state"].PkColumns.ShouldBe(["scripted_alarm_id"]); } + [Fact] + public void AlarmConditionComment_PkIsTheAlarmIdPlusTheTrailOrdinal() + { + // One row per trail entry. Keyed by ordinal (not content) so the pair's two copies of the + // same logical entry — each stamped by its own node's clock — collapse to one row under LWW. + var db = Boot(InitiatorReplication); + + db.ReplicatedTables["alarm_condition_comment"].PkColumns.ShouldBe(["scripted_alarm_id", "seq"]); + } + + [Fact] + public async Task OnReady_MovesAnOldShapeInlineCommentTrail_AfterRegistration_SoTheMoveReplicates() + { + // A file written by the pre-split store: the trail is inline on the condition row. + var first = Boot(InitiatorReplication); + await first.ExecuteAsync( + """ + INSERT INTO alarm_condition_state + (scripted_alarm_id, enabled_state, acked_state, confirmed_state, shelving_state, + comments_json, updated_at_utc) + VALUES ('eq/alarm-legacy', 'Enabled', 'Acknowledged', 'Confirmed', 'Unshelved', + '[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"Acknowledge","text":"a"},{"timestampUtc":"2026-06-10T10:05:00Z","user":"bob","kind":"Confirm","text":"b"}]', + '2026-06-10T10:05:00.0000000Z') + """, + parameters: null, + TestContext.Current.CancellationToken); + // Positive control: before the restart the trail is inline and the comment table is empty. + (await ScalarAsync(first, "SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(0); + + // The upgrade restart. + var db = Boot(InitiatorReplication); + + (await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'eq/alarm-legacy'")) + .ShouldBe(2); + (await ScalarAsync(db, "SELECT COUNT(*) FROM alarm_condition_state WHERE comments_json <> '[]'")) + .ShouldBe(0); + // Captured: the migration ran after RegisterReplicated, so the moved entries reach the peer. + (await ScalarAsync(db, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'")) + .ShouldBe(2); + + // Idempotent: a further boot writes nothing more. + var third = Boot(InitiatorReplication); + (await ScalarAsync(third, "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'")) + .ShouldBe(2); + (await ScalarAsync(third, "SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(2); + } + [Fact] public async Task AlarmSfEventRows_EnterTheOplog() { @@ -296,7 +343,7 @@ public sealed class LocalDbSetupTests : IDisposable var replicated = Boot(InitiatorReplication); await InsertPointerAsync(replicated, "SITE-A"); (await ScalarAsync(replicated, "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger'")) - .ShouldBe(12); + .ShouldBe(15); var unreplicated = Boot(); @@ -365,13 +412,13 @@ public sealed class LocalDbSetupTests : IDisposable [Fact] public void AFailureMidwayThroughRegistration_FailsBootLoudly_AndTheRetriedBootConverges() { - // OnReady flips four tables in a loop with no try/catch, so a failure on the third leaves - // two registered and two not. That is the RIGHT behaviour and this pins both halves of why: + // OnReady flips five tables in a loop with no try/catch, so a failure on the third leaves + // two registered and three not. That is the RIGHT behaviour and this pins both halves of why: // the throw propagates out of the AddZbLocalDb singleton factory and fails host start // (a node half-capturing is a node whose replication is silently partial), and because both // library APIs are idempotent the retried boot completes the flip rather than compounding it. // - // Swallowing the throw would leave the node up, serving, and replicating two of four tables + // Swallowing the throw would leave the node up, serving, and replicating two of five tables // with nothing in the log to say so. var real = Boot(InitiatorReplication); @@ -387,11 +434,11 @@ public sealed class LocalDbSetupTests : IDisposable // pinning some unrelated throw, and the "partial flip" it claims to model never happened. faulty.Calls.ShouldBe(3); - // The retry: same file, no wrapper. Idempotent re-registration converges on all four. + // The retry: same file, no wrapper. Idempotent re-registration converges on all five. LocalDbSetup.OnReady(real, configuration); real.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs index 6798faf5..5477afea 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDbWiringTests.cs @@ -108,7 +108,7 @@ public sealed class LocalDbWiringTests : IDisposable // Exact set, both directions — a dropped registration replicates nothing; an extra one costs // three triggers and oplog volume on every write. first.ReplicatedTables.Keys.OrderBy(k => k, StringComparer.Ordinal) - .ShouldBe(["alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); + .ShouldBe(["alarm_condition_comment", "alarm_condition_state", "alarm_sf_events", "deployment_artifacts", "deployment_pointer"]); } [Fact] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs new file mode 100644 index 00000000..47438ceb --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmCommentAuditTests.cs @@ -0,0 +1,383 @@ +using System.Collections.Immutable; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; +using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.ScriptedAlarms; + +/// +/// The alarm comment audit trail lives in its own append-only table +/// () keyed by alarm id, so the condition +/// row stays a fixed size and a transition's write — and therefore its replication payload — +/// does not grow with the alarm's history. +/// +/// +/// The payload is measured where it costs: the bytes the LocalDb capture triggers write into +/// __localdb_oplog for one save, which is exactly what ships to the pair peer. Both +/// tables are registered for capture here, as they are on a replicating node. +/// +public sealed class LocalDbAlarmCommentAuditTests : IDisposable +{ + private static readonly DateTime T0 = new(2026, 06, 10, 12, 00, 00, DateTimeKind.Utc); + + private readonly string _dbPath = + Path.Combine(Path.GetTempPath(), $"otopcua-alarm-comments-{Guid.NewGuid():N}.db"); + + private readonly ServiceProvider _provider; + private readonly ILocalDb _db; + + public LocalDbAlarmCommentAuditTests() + { + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = _dbPath }) + .Build(); + + _provider = new ServiceCollection() + .AddZbLocalDb(configuration, db => + { + using (var connection = db.CreateConnection()) + { + AlarmConditionStateSchema.Apply(connection); + } + + db.RegisterReplicated(AlarmConditionStateSchema.StateTable); + db.RegisterReplicated(AlarmConditionStateSchema.CommentsTable); + }) + .BuildServiceProvider(); + + _db = _provider.GetRequiredService(); + } + + public void Dispose() + { + _provider.Dispose(); + SqliteConnection.ClearAllPools(); + foreach (var path in new[] { _dbPath, $"{_dbPath}-wal", $"{_dbPath}-shm" }) + { + try { if (File.Exists(path)) File.Delete(path); } catch { /* best effort */ } + } + } + + private LocalDbAlarmConditionStateStore NewStore() + => new(_db, NullLogger.Instance); + + private static CancellationToken Ct => CancellationToken.None; + + private static ImmutableList Comments(int count, string prefix = "c") + { + var builder = ImmutableList.CreateBuilder(); + for (var i = 0; i < count; i++) + builder.Add(new AlarmComment(T0.AddSeconds(i), "jane", "AddComment", $"{prefix}-{i:D6}")); + return builder.ToImmutable(); + } + + /// An unacknowledged state carrying , ready to be acked. + private static AlarmConditionState Unacked(string alarmId, ImmutableList comments) => + AlarmConditionState.Fresh(alarmId, T0) with + { + Active = AlarmActiveState.Active, + Acked = AlarmAckedState.Unacknowledged, + Comments = comments, + }; + + private async Task OplogHighWaterAsync() => + (await _db.QueryAsync("SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog", r => r.GetInt64(0), null, Ct))[0]; + + /// Bytes captured into the oplog above — the replication payload. + private async Task OplogBytesAboveAsync(long afterSeq) => + (await _db.QueryAsync( + "SELECT COALESCE(SUM(LENGTH(pk_json) + COALESCE(LENGTH(row_json), 0)), 0) FROM __localdb_oplog WHERE seq > @S", + r => r.GetInt64(0), + new { S = afterSeq }, + Ct))[0]; + + private async Task OplogEntriesAboveAsync(long afterSeq, string table) => + (await _db.QueryAsync( + "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > @S AND table_name = @T", + r => r.GetInt64(0), + new { S = afterSeq, T = table }, + Ct))[0]; + + private async Task ScalarAsync(string sql) => + (await _db.QueryAsync(sql, r => r.GetInt64(0), null, Ct))[0]; + + /// + /// Seeds with comments, then + /// returns the oplog bytes one acknowledge transition (one new comment) costs. + /// + private async Task AckTransitionBytesAsync(string alarmId, int historyLength, string ackComment) + { + var store = NewStore(); + var seeded = Unacked(alarmId, Comments(historyLength)); + await store.SaveAsync(seeded, Ct); + + var before = await OplogHighWaterAsync(); + var acked = Part9StateMachine.ApplyAcknowledge(seeded, "bob", ackComment, T0.AddDays(1)).State; + await store.SaveAsync(acked, Ct); + return await OplogBytesAboveAsync(before); + } + + [Fact] + public async Task TransitionPayload_IsConstant_WhetherTheAlarmCarries10OrTenThousandComments() + { + var small = await AckTransitionBytesAsync("alarm-small", 10, "ack"); + var large = await AckTransitionBytesAsync("alarm-large", 10_000, "ack"); + + // Constant: the only difference allowed is the width of the new comment's ordinal ("10" vs + // "10000", in pk_json and row_json) — a handful of bytes, not O(n). The alarm ids are the + // same length on purpose. + small.ShouldBeGreaterThan(0); + Math.Abs(large - small).ShouldBeLessThanOrEqualTo(16, + $"one ack transition cost {small} bytes at 10 comments but {large} at 10,000"); + large.ShouldBeLessThan(2_048); + + // Positive control on the instrument: the seeded history really is large and really was + // captured — otherwise a "constant" result could just mean capture is off. + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-large'")) + .ShouldBe(10_001); + var historyBytes = await ScalarAsync( + "SELECT SUM(LENGTH(row_json)) FROM __localdb_oplog WHERE table_name = 'alarm_condition_comment'"); + historyBytes.ShouldBeGreaterThan(1_000_000); + } + + [Fact] + public async Task TransitionPayload_PositiveControl_TheMeasurementSeesTheNewCommentsBytes() + { + // Proves the measurement above is not vacuous: the oplog delta of a transition does carry + // the appended comment, so a 4 KB comment costs at least 4 KB. + var plain = await AckTransitionBytesAsync("alarm-plain", 10, "ack"); + var heavy = await AckTransitionBytesAsync("alarm-heavy", 10, new string('x', 4_096)); + + (heavy - plain).ShouldBeGreaterThanOrEqualTo(4_096 * 2 - 16, + "the comment text rides both the comment row and last_ack_comment on the condition row"); + } + + [Fact] + public async Task Transition_WritesOneConditionRowAndOnlyTheAppendedCommentRow() + { + var store = NewStore(); + var seeded = Unacked("alarm-append", Comments(50)); + await store.SaveAsync(seeded, Ct); + + var before = await OplogHighWaterAsync(); + var acked = Part9StateMachine.ApplyAcknowledge(seeded, "bob", "ack", T0.AddDays(1)).State; + await store.SaveAsync(acked, Ct); + + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(1); + + var loaded = await store.LoadAsync("alarm-append", Ct); + loaded.ShouldNotBeNull(); + loaded.Comments.ShouldBe(acked.Comments); + } + + [Fact] + public async Task ConditionRow_StaysFixedSize_InlineColumnIsAlwaysEmpty() + { + var store = NewStore(); + await store.SaveAsync(Unacked("alarm-fixed", Comments(500)), Ct); + + (await _db.QueryAsync( + "SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-fixed'", + r => r.GetString(0), null, Ct))[0] + .ShouldBe("[]"); + (await store.LoadAsync("alarm-fixed", Ct))!.Comments.Count.ShouldBe(500); + } + + [Fact] + public async Task UnchangedSave_WritesNoCommentRows() + { + var store = NewStore(); + var state = Unacked("alarm-idle", Comments(20)); + await store.SaveAsync(state, Ct); + + var before = await OplogHighWaterAsync(); + await store.SaveAsync(state with { LastTransitionUtc = T0.AddMinutes(1) }, Ct); + + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.CommentsTable)).ShouldBe(0); + (await OplogEntriesAboveAsync(before, AlarmConditionStateSchema.StateTable)).ShouldBe(1, + "positive control: the save itself was captured"); + } + + [Fact] + public async Task LoadAll_ReturnsEachAlarmsOwnTrailInOrder() + { + var store = NewStore(); + var a = Unacked("alarm-a", Comments(3, "a")); + var b = Unacked("alarm-b", Comments(5, "b")); + await store.SaveAsync(a, Ct); + await store.SaveAsync(b, Ct); + + var all = (await store.LoadAllAsync(Ct)).ToDictionary(s => s.AlarmId); + + all["alarm-a"].Comments.ShouldBe(a.Comments); + all["alarm-b"].Comments.ShouldBe(b.Comments); + } + + [Fact] + public async Task Remove_DeletesTheAlarmsTrail_AndAReCreatedAlarmStartsEmpty() + { + var store = NewStore(); + await store.SaveAsync(Unacked("alarm-gone", Comments(4)), Ct); + await store.SaveAsync(Unacked("alarm-kept", Comments(2)), Ct); + + await store.RemoveAsync("alarm-gone", Ct); + + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-gone'")) + .ShouldBe(0); + (await store.LoadAsync("alarm-kept", Ct))!.Comments.Count.ShouldBe(2, "positive control"); + + var recreated = AlarmConditionState.Fresh("alarm-gone", T0) + with { Comments = ImmutableList.Create(new AlarmComment(T0, "amy", "AddComment", "new life")) }; + await store.SaveAsync(recreated, Ct); + (await store.LoadAsync("alarm-gone", Ct))!.Comments.ShouldBe(recreated.Comments); + } + + [Fact] + public async Task CommentsAlreadyHeldAtAnOrdinal_AreNeverRewritten() + { + // Append-only: a save whose list disagrees with what is stored at an existing ordinal (the + // peer's copy of the same logical entry already arrived) leaves the stored entry alone. + var store = NewStore(); + var original = Unacked("alarm-ao", Comments(3)); + await store.SaveAsync(original, Ct); + + var rewritten = original with { Comments = original.Comments.SetItem(1, new AlarmComment(T0, "eve", "AddComment", "forged")) }; + await store.SaveAsync(rewritten, Ct); + + (await store.LoadAsync("alarm-ao", Ct))!.Comments.ShouldBe(original.Comments); + } + + // ---- back-compat + migration ----------------------------------------------------------- + + /// + /// The exact shape the pre-split store wrote into comments_json + /// (JsonSerializerDefaults.Web: camelCase, Utc timestamps with a trailing Z). + /// + private const string OldShapeCommentsJson = + """[{"timestampUtc":"2026-06-10T10:00:00Z","user":"jane","kind":"Acknowledge","text":"checking pump"},{"timestampUtc":"2026-06-10T11:30:00.1234567Z","user":"bob","kind":"Confirm","text":"all clear"},{"timestampUtc":"2026-06-10T11:30:00.1234567Z","user":"system","kind":"AutoUnshelve","text":""}]"""; + + private static readonly ImmutableList OldShapeComments = ImmutableList.Create( + new AlarmComment(new DateTime(2026, 06, 10, 10, 00, 00, DateTimeKind.Utc), "jane", "Acknowledge", "checking pump"), + new AlarmComment(new DateTime(2026, 06, 10, 11, 30, 00, DateTimeKind.Utc).AddTicks(1_234_567), "bob", "Confirm", "all clear"), + new AlarmComment(new DateTime(2026, 06, 10, 11, 30, 00, DateTimeKind.Utc).AddTicks(1_234_567), "system", "AutoUnshelve", "")); + + private async Task InsertOldShapeRowAsync(string alarmId, string commentsJson = OldShapeCommentsJson) => + 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', 'Acknowledged', 'Confirmed', 'Unshelved', + NULL, 'jane', 'checking pump', '2026-06-10T10:00:00.0000000Z', + 'bob', 'all clear', '2026-06-10T11:30:00.1234567Z', @Json, '2026-06-10T11:30:00.1234567Z') + """, + new { Id = alarmId, Json = commentsJson }, + Ct); + + private static void ShouldBeTheOldShapeState(AlarmConditionState? loaded, string alarmId) + { + loaded.ShouldNotBeNull(); + loaded.AlarmId.ShouldBe(alarmId); + loaded.Acked.ShouldBe(AlarmAckedState.Acknowledged); + loaded.Confirmed.ShouldBe(AlarmConfirmedState.Confirmed); + loaded.LastAckUser.ShouldBe("jane"); + loaded.LastConfirmComment.ShouldBe("all clear"); + loaded.Comments.ShouldBe(OldShapeComments); + loaded.Comments.ShouldAllBe(c => c.TimestampUtc.Kind == DateTimeKind.Utc); + } + + [Fact] + public async Task OldShapeRow_WithInlineComments_LoadsExactlyAsBefore() + { + await InsertOldShapeRowAsync("alarm-legacy"); + + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-legacy", Ct), "alarm-legacy"); + ShouldBeTheOldShapeState((await NewStore().LoadAllAsync(Ct)).Single(), "alarm-legacy"); + } + + [Fact] + public async Task Migration_MovesInlineCommentsToTheAuditTable_AndClearsTheInlineColumn() + { + await InsertOldShapeRowAsync("alarm-m1"); + await InsertOldShapeRowAsync("alarm-m2"); + await InsertOldShapeRowAsync("alarm-empty", "[]"); + + int migrated; + using (var connection = _db.CreateConnection()) + migrated = AlarmConditionStateSchema.MigrateInlineComments(connection); + + migrated.ShouldBe(2); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(6); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_state WHERE comments_json <> '[]'")).ShouldBe(0); + + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-m1", Ct), "alarm-m1"); + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-m2", Ct), "alarm-m2"); + (await NewStore().LoadAsync("alarm-empty", Ct))!.Comments.ShouldBeEmpty(); + } + + [Fact] + public async Task Migration_IsIdempotent_ASecondRunChangesNothing() + { + await InsertOldShapeRowAsync("alarm-idem"); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).ShouldBe(1, "positive control"); + + var before = await OplogHighWaterAsync(); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).ShouldBe(0); + + (await OplogHighWaterAsync()).ShouldBe(before, "a re-run must write nothing"); + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(3); + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-idem", Ct), "alarm-idem"); + } + + [Fact] + public async Task Migration_OverAPartiallyMigratedRow_DoesNotDuplicateOrReorder() + { + // An inline row reappearing after its comments were already moved (e.g. a restored backup): + // the ordinals already held are kept, nothing is duplicated. + await InsertOldShapeRowAsync("alarm-partial"); + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection); + await _db.ExecuteAsync( + "UPDATE alarm_condition_state SET comments_json = @Json WHERE scripted_alarm_id = 'alarm-partial'", + new { Json = OldShapeCommentsJson }, Ct); + + using (var connection = _db.CreateConnection()) + AlarmConditionStateSchema.MigrateInlineComments(connection).ShouldBe(1); + + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment")).ShouldBe(3); + ShouldBeTheOldShapeState(await NewStore().LoadAsync("alarm-partial", Ct), "alarm-partial"); + } + + [Fact] + public async Task SavingALoadedOldShapeState_MovesItsCommentsAndAppendsTheNewOne() + { + // The lazy path: an old-shape row the boot migration never saw (it was written before the + // first save on this build) still converges on the first save. + await InsertOldShapeRowAsync("alarm-lazy"); + var store = NewStore(); + var loaded = await store.LoadAsync("alarm-lazy", Ct); + + var next = Part9StateMachine.ApplyAddComment(loaded!, "amy", "follow-up", T0.AddDays(2)).State; + await store.SaveAsync(next, Ct); + + (await ScalarAsync("SELECT COUNT(*) FROM alarm_condition_comment WHERE scripted_alarm_id = 'alarm-lazy'")) + .ShouldBe(4); + (await _db.QueryAsync( + "SELECT comments_json FROM alarm_condition_state WHERE scripted_alarm_id = 'alarm-lazy'", + r => r.GetString(0), null, Ct))[0] + .ShouldBe("[]"); + (await NewStore().LoadAsync("alarm-lazy", Ct))!.Comments.ShouldBe(next.Comments); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs index fc0d9525..20a4085e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ScriptedAlarms/LocalDbAlarmConditionStateStoreTests.cs @@ -48,6 +48,7 @@ public sealed class LocalDbAlarmConditionStateStoreTests : IDisposable } db.RegisterReplicated(AlarmConditionStateSchema.StateTable); + db.RegisterReplicated(AlarmConditionStateSchema.CommentsTable); }) .BuildServiceProvider();