fix(store-and-forward): standby applies Add/Park/Requeue as upserts so a lost Add self-heals

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

@@ -101,6 +101,13 @@ public class ReplicationService
     /// <summary>
     /// Applies a replicated operation received from the active node.
     /// Used by the standby node to keep its SQLite in sync.
+    ///
+    /// Add/Park/Requeue are applied as <b>upserts</b> (<see cref="StoreAndForwardStorage.UpsertMessageAsync"/>),
+    /// not blind INSERT/UPDATE: the full message rides in every one of those operations,
+    /// so a Park/Requeue whose original Add was lost (fire-and-forget replication is
+    /// best-effort) self-heals by materialising the row, and a duplicate Add (e.g.
+    /// re-issued by an anti-entropy resync) applies newest-wins instead of throwing a
+    /// primary-key violation. Remove is a plain delete.
     /// </summary>
     /// <param name="operation">The replication operation to apply.</param>
     /// <param name="storage">The standby node's store-and-forward storage to update.</param>
@@ -112,7 +119,7 @@ public class ReplicationService
         switch (operation.OperationType)
         {
             case ReplicationOperationType.Add when operation.Message != null:
-                await storage.EnqueueAsync(operation.Message);
+                await storage.UpsertMessageAsync(operation.Message);
                 break;
 
             case ReplicationOperationType.Remove:
@@ -121,13 +128,13 @@ public class ReplicationService
 
             case ReplicationOperationType.Park when operation.Message != null:
                 operation.Message.Status = StoreAndForwardMessageStatus.Parked;
-                await storage.UpdateMessageAsync(operation.Message);
+                await storage.UpsertMessageAsync(operation.Message);
                 break;
 
             case ReplicationOperationType.Requeue when operation.Message != null:
                 operation.Message.Status = StoreAndForwardMessageStatus.Pending;
                 operation.Message.RetryCount = 0;
-                await storage.UpdateMessageAsync(operation.Message);
+                await storage.UpsertMessageAsync(operation.Message);
                 break;
         }
     }

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

@@ -176,6 +176,67 @@ public class StoreAndForwardStorage
         await cmd.ExecuteNonQueryAsync();
     }
 
+    /// <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
+    /// node when applying a replicated Add/Park/Requeue: a Park/Requeue whose original
+    /// Add was lost self-heals (the full message rides in the operation), and a
+    /// duplicate Add (e.g. after an anti-entropy resync) applies newest-wins instead of
+    /// violating the primary key. The original <c>created_at</c> is preserved — it
+    /// orders the retry sweep and must never be overwritten on update.
+    /// </summary>
+    /// <param name="message">The message to insert or update.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    public async Task UpsertMessageAsync(StoreAndForwardMessage message)
+    {
+        await using var connection = new SqliteConnection(_connectionString);
+        await connection.OpenAsync();
+
+        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, status, last_error,
+                                     origin_instance, execution_id, source_script, parent_execution_id)
+            VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries,
+                    @retryIntervalMs, @createdAt, @lastAttempt, @status, @lastError,
+                    @origin, @executionId, @sourceScript, @parentExecutionId)
+            ON CONFLICT(id) DO UPDATE SET
+                category = excluded.category,
+                target = excluded.target,
+                payload_json = excluded.payload_json,
+                retry_count = excluded.retry_count,
+                max_retries = excluded.max_retries,
+                retry_interval_ms = excluded.retry_interval_ms,
+                last_attempt_at = excluded.last_attempt_at,
+                status = excluded.status,
+                last_error = excluded.last_error,
+                origin_instance = excluded.origin_instance,
+                execution_id = excluded.execution_id,
+                source_script = excluded.source_script,
+                parent_execution_id = excluded.parent_execution_id";
+
+        cmd.Parameters.AddWithValue("@id", message.Id);
+        cmd.Parameters.AddWithValue("@category", (int)message.Category);
+        cmd.Parameters.AddWithValue("@target", message.Target);
+        cmd.Parameters.AddWithValue("@payload", message.PayloadJson);
+        cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
+        cmd.Parameters.AddWithValue("@maxRetries", message.MaxRetries);
+        cmd.Parameters.AddWithValue("@retryIntervalMs", message.RetryIntervalMs);
+        cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.ToString("O"));
+        cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
+            ? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
+        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);
+        cmd.Parameters.AddWithValue("@executionId",
+            message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value);
+        cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value);
+        cmd.Parameters.AddWithValue("@parentExecutionId",
+            message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value);
+
+        await cmd.ExecuteNonQueryAsync();
+    }
+
     /// <summary>
     /// Gets all messages that are due for retry (Pending status, last attempt older than retry interval).
     /// </summary>

tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs

@@ -203,4 +203,61 @@ public class StoreAndForwardReplicationTests : IAsyncLifetime, IDisposable
         Assert.Equal(StoreAndForwardMessageStatus.Pending, row!.Status);
         Assert.Equal(0, row.RetryCount);
     }
+
+    private static ReplicationService NewReplicationService() =>
+        new(new StoreAndForwardOptions { ReplicationEnabled = true },
+            NullLogger<ReplicationService>.Instance);
+
+    private static StoreAndForwardMessage NewMessage(string id) => new()
+    {
+        Id = id,
+        Category = StoreAndForwardCategory.ExternalSystem,
+        Target = "api",
+        PayloadJson = "{}",
+        RetryCount = 0,
+        MaxRetries = 1,
+        RetryIntervalMs = 0,
+        CreatedAt = DateTimeOffset.UtcNow,
+        Status = StoreAndForwardMessageStatus.Pending,
+    };
+
+    /// <summary>
+    /// A Park (or Requeue) whose original Add was lost — e.g. the Add's
+    /// fire-and-forget replication dropped — must still materialise the row on the
+    /// standby. The full message rides in the operation, so an upsert self-heals;
+    /// a blind UPDATE would affect 0 rows and the row would be gone forever.
+    /// </summary>
+    [Fact]
+    public async Task ApplyReplicatedPark_WhenAddWasLost_MaterializesTheParkedRow()
+    {
+        var service = NewReplicationService();
+        var msg = NewMessage("lost-add-1"); // never Added on this (standby) storage
+        msg.Status = StoreAndForwardMessageStatus.Parked;
+
+        await service.ApplyReplicatedOperationAsync(
+            new ReplicationOperation(ReplicationOperationType.Park, msg.Id, msg), _storage);
+
+        var row = await _storage.GetMessageByIdAsync("lost-add-1");
+        Assert.NotNull(row); // pre-fix: blind UPDATE affected 0 rows, row is gone forever
+        Assert.Equal(StoreAndForwardMessageStatus.Parked, row!.Status);
+    }
+
+    /// <summary>
+    /// A duplicate Add (e.g. after Task 21's anti-entropy resync re-issues an Add
+    /// the standby already holds) must not violate the PK — the upsert applies
+    /// newest-wins instead of throwing.
+    /// </summary>
+    [Fact]
+    public async Task ApplyReplicatedAdd_Twice_IsIdempotent_NewestWins()
+    {
+        var service = NewReplicationService();
+        var msg = NewMessage("dup-add");
+        await service.ApplyReplicatedOperationAsync(
+            new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage);
+        msg.RetryCount = 3;
+        await service.ApplyReplicatedOperationAsync( // pre-fix: SqliteException PK violation
+            new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage);
+
+        Assert.Equal(3, (await _storage.GetMessageByIdAsync("dup-add"))!.RetryCount);
+    }
 }