diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
index cfbe34ea..26a7a7d4 100644
--- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
@@ -185,16 +185,11 @@ public class StoreAndForwardStorage
}
///
- /// Enqueues a new message with Pending status.
+ /// INSERT statement for a full message row. Shared by
+ /// and ; bind with
+ /// so the column list and the parameters never drift apart.
///
- /// The message to enqueue.
- /// A task that represents the asynchronous operation.
- public async Task EnqueueAsync(StoreAndForwardMessage message)
- {
- await using var connection = await OpenConnectionAsync();
-
- await using var cmd = connection.CreateCommand();
- cmd.CommandText = @"
+ private const string InsertMessageSql = @"
INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries,
retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error,
origin_instance, execution_id, source_script, parent_execution_id)
@@ -202,6 +197,14 @@ public class StoreAndForwardStorage
@retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError,
@origin, @executionId, @sourceScript, @parentExecutionId)";
+ ///
+ /// Binds every column parameter for a full message row (used by
+ /// ). GUID ids are stored in canonical "D" string
+ /// form and the epoch-ms sibling of last_attempt_at is derived here so a single
+ /// place owns the mapping.
+ ///
+ private static void BindMessageParameters(SqliteCommand cmd, StoreAndForwardMessage message)
+ {
cmd.Parameters.AddWithValue("@id", message.Id);
cmd.Parameters.AddWithValue("@category", (int)message.Category);
cmd.Parameters.AddWithValue("@target", message.Target);
@@ -217,21 +220,91 @@ public class StoreAndForwardStorage
cmd.Parameters.AddWithValue("@status", (int)message.Status);
cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
cmd.Parameters.AddWithValue("@origin", (object?)message.OriginInstanceName ?? DBNull.Value);
- // The execution id is stored as its
- // canonical string form ("D") so it round-trips cleanly through the
- // TEXT column; null when not a cached call / not threaded.
+ // GUID ids stored as canonical "D" strings; null when not threaded.
cmd.Parameters.AddWithValue("@executionId",
message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value);
cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value);
- // The parent execution id is
- // stored as its canonical string form ("D") so it round-trips cleanly
- // through the TEXT column; null when not a routed cached call.
cmd.Parameters.AddWithValue("@parentExecutionId",
message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value);
+ }
+
+ ///
+ /// Enqueues a new message with Pending status.
+ ///
+ /// The message to enqueue.
+ /// A task that represents the asynchronous operation.
+ public async Task EnqueueAsync(StoreAndForwardMessage message)
+ {
+ await using var connection = await OpenConnectionAsync();
+
+ await using var cmd = connection.CreateCommand();
+ cmd.CommandText = InsertMessageSql;
+ BindMessageParameters(cmd, message);
await cmd.ExecuteNonQueryAsync();
}
+ ///
+ /// Returns every buffered message regardless of status, oldest-first, up to
+ /// rows, plus a flag indicating the buffer holds more
+ /// than that. Anti-entropy groundwork: the active node snapshots its buffer for
+ /// a standby resync. One extra row is fetched (limit + 1) purely to
+ /// compute the truncation flag without a second COUNT round-trip.
+ ///
+ /// Maximum rows to return; must be positive.
+ /// The oldest-first page (at most rows) and whether more rows exist beyond it.
+ public async Task<(List Messages, bool Truncated)> GetAllMessagesAsync(int limit)
+ {
+ await using var connection = await OpenConnectionAsync();
+
+ await using var cmd = connection.CreateCommand();
+ cmd.CommandText = @"
+ SELECT id, category, target, payload_json, retry_count, max_retries,
+ retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
+ execution_id, source_script, parent_execution_id
+ FROM sf_messages
+ ORDER BY created_at ASC
+ LIMIT @limitPlusOne";
+ cmd.Parameters.AddWithValue("@limitPlusOne", limit + 1);
+
+ var rows = await ReadMessagesAsync(cmd);
+ var truncated = rows.Count > limit;
+ return (rows.Take(limit).ToList(), truncated);
+ }
+
+ ///
+ /// Atomically replaces the entire buffer with in a
+ /// single transaction (delete-all then insert-all). Standby-side anti-entropy
+ /// apply: a peer-join resync overwrites the standby's divergent copy with the
+ /// active node's authoritative snapshot. Never call on an active node —
+ /// it discards every in-flight row.
+ ///
+ /// The full buffer snapshot to install.
+ /// A task that represents the asynchronous operation.
+ public async Task ReplaceAllAsync(IReadOnlyList messages)
+ {
+ await using var connection = await OpenConnectionAsync();
+ await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
+
+ await using (var deleteCmd = connection.CreateCommand())
+ {
+ deleteCmd.Transaction = transaction;
+ deleteCmd.CommandText = "DELETE FROM sf_messages";
+ await deleteCmd.ExecuteNonQueryAsync();
+ }
+
+ foreach (var message in messages)
+ {
+ await using var insertCmd = connection.CreateCommand();
+ insertCmd.Transaction = transaction;
+ insertCmd.CommandText = InsertMessageSql;
+ BindMessageParameters(insertCmd, message);
+ await insertCmd.ExecuteNonQueryAsync();
+ }
+
+ await transaction.CommitAsync();
+ }
+
///
/// Inserts a message, or updates every mutable column in place if a row with the
/// same id already exists (ON CONFLICT(id) DO UPDATE). Used by the standby
diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs
index 3ee14b0f..317d704c 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardStorageTests.cs
@@ -748,4 +748,46 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable
var result = await cmd.ExecuteScalarAsync();
return result is DBNull ? null : result;
}
+
+ // ── Task 20: buffer-resync storage primitives ──
+
+ private static StoreAndForwardMessage NewMsg(
+ string id,
+ StoreAndForwardMessageStatus status = StoreAndForwardMessageStatus.Pending,
+ DateTimeOffset? createdAt = null) => new()
+ {
+ Id = id,
+ Category = StoreAndForwardCategory.ExternalSystem,
+ Target = "t",
+ PayloadJson = "{}",
+ RetryCount = 0,
+ MaxRetries = 50,
+ RetryIntervalMs = 30000,
+ CreatedAt = createdAt ?? DateTimeOffset.UtcNow,
+ Status = status,
+ };
+
+ [Fact]
+ public async Task GetAllMessages_ReturnsEveryStatus_OldestFirst_UpToLimit()
+ {
+ var @base = DateTimeOffset.UtcNow;
+ await _storage.EnqueueAsync(NewMsg("p1", StoreAndForwardMessageStatus.Pending, @base));
+ await _storage.EnqueueAsync(NewMsg("k1", StoreAndForwardMessageStatus.Parked, @base.AddSeconds(1)));
+ await _storage.EnqueueAsync(NewMsg("p2", StoreAndForwardMessageStatus.Pending, @base.AddSeconds(2)));
+
+ var (all, truncated) = await _storage.GetAllMessagesAsync(limit: 2);
+ Assert.Equal(new[] { "p1", "k1" }, all.Select(m => m.Id)); // every status, oldest-first
+ Assert.True(truncated); // a third row exists beyond the limit
+ }
+
+ [Fact]
+ public async Task ReplaceAll_SwapsTheEntireBuffer_Atomically()
+ {
+ await _storage.EnqueueAsync(NewMsg("stale"));
+ await _storage.ReplaceAllAsync(new[] { NewMsg("fresh-1"), NewMsg("fresh-2") });
+
+ Assert.Null(await _storage.GetMessageByIdAsync("stale"));
+ Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-1"));
+ Assert.NotNull(await _storage.GetMessageByIdAsync("fresh-2"));
+ }
}