feat(store-and-forward): GetAllMessagesAsync/ReplaceAllAsync storage primitives for standby buffer resync

src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs

@@ -185,16 +185,11 @@ public class StoreAndForwardStorage
     }
 
     /// <summary>
-    /// Enqueues a new message with Pending status.
+    /// INSERT statement for a full message row. Shared by <see cref="EnqueueAsync"/>
+    /// and <see cref="ReplaceAllAsync"/>; bind with <see cref="BindMessageParameters"/>
+    /// so the column list and the parameters never drift apart.
     /// </summary>
-    /// <param name="message">The message to enqueue.</param>
+    private const string InsertMessageSql = @"
-    /// <returns>A task that represents the asynchronous operation.</returns>
-    public async Task EnqueueAsync(StoreAndForwardMessage message)
-    {
-        await using var connection = await OpenConnectionAsync();
-
-        await using var cmd = connection.CreateCommand();
-        cmd.CommandText = @"
             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)";
 
+    /// <summary>
+    /// Binds every column parameter for a full message row (used by
+    /// <see cref="InsertMessageSql"/>). 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.
+    /// </summary>
+    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
+        // GUID ids stored as canonical "D" strings; null when not threaded.
-        // canonical string form ("D") so it round-trips cleanly through the
-        // TEXT column; null when not a cached call / 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);
+    }
+
+    /// <summary>
+    /// Enqueues a new message with Pending status.
+    /// </summary>
+    /// <param name="message">The message to enqueue.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    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();
     }
 
+    /// <summary>
+    /// Returns every buffered message regardless of status, oldest-first, up to
+    /// <paramref name="limit"/> 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 (<c>limit + 1</c>) purely to
+    /// compute the truncation flag without a second COUNT round-trip.
+    /// </summary>
+    /// <param name="limit">Maximum rows to return; must be positive.</param>
+    /// <returns>The oldest-first page (at most <paramref name="limit"/> rows) and whether more rows exist beyond it.</returns>
+    public async Task<(List<StoreAndForwardMessage> 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);
+    }
+
+    /// <summary>
+    /// Atomically replaces the entire buffer with <paramref name="messages"/> 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. <b>Never call on an active node</b> —
+    /// it discards every in-flight row.
+    /// </summary>
+    /// <param name="messages">The full buffer snapshot to install.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    public async Task ReplaceAllAsync(IReadOnlyList<StoreAndForwardMessage> 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();
+    }
+
     /// <summary>
     /// Inserts a message, or updates every mutable column in place if a row with the
     /// same id already exists (<c>ON CONFLICT(id) DO UPDATE</c>). Used by the standby

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"));
+    }
 }