Implemented SetEffectResults with remove effects overload

stidsborg · stidsborg · commit a2b25e1e3296 · 2025-02-07T15:31:28.000+01:00
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/EffectStoreTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/EffectStoreTests.cs
@@ -34,4 +34,8 @@ public override Task TruncateDeletesAllEffects()
     [TestMethod]
     public override Task BulkInsertTest()
         => BulkInsertTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
+
+    [TestMethod]
+    public override Task BulkInsertAndDeleteTest()
+        => BulkInsertAndDeleteTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
 }
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/EffectStoreTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/EffectStoreTests.cs
@@ -278,4 +278,44 @@ protected async Task BulkInsertTest(Task<IEffectsStore> storeTask)
         var effect2 = effects.Single(e => e.EffectId == storedEffect2.EffectId);
         effect2.Result.ShouldBe("some result 2".ToUtf8Bytes());
     }
+    
+    public abstract Task BulkInsertAndDeleteTest();
+    protected async Task BulkInsertAndDeleteTest(Task<IEffectsStore> storeTask)
+    {
+        var store = await storeTask;
+        var storedId = TestStoredId.Create();
+        var storedEffect1 = new StoredEffect(
+            "EffectId1".ToEffectId(),
+            "EffectId1".ToStoredEffectId(EffectType.Effect),
+            WorkStatus.Started,
+            Result: "some result 1".ToUtf8Bytes(),
+            StoredException: null
+        );
+        var storedEffect2 = new StoredEffect(
+            "EffectId2".ToEffectId(),
+            "EffectId2".ToStoredEffectId(EffectType.Effect),
+            WorkStatus.Completed,
+            Result: "some result 2".ToUtf8Bytes(),
+            StoredException: null
+        );
+        var storedEffect3 = new StoredEffect(
+            "EffectId3".ToEffectId(),
+            "EffectId3".ToStoredEffectId(EffectType.Effect),
+            WorkStatus.Completed,
+            Result: "some result 3".ToUtf8Bytes(),
+            StoredException: null
+        );
+        
+        await store.SetEffectResults(storedId, [storedEffect1, storedEffect2]);
+        await store.SetEffectResults(
+            storedId,
+            upsertEffects: [storedEffect3],
+            removeEffects: [storedEffect1.StoredEffectId, storedEffect2.StoredEffectId]
+        );
+        
+        var effects = await store.GetEffectResults(storedId);
+        effects.Count.ShouldBe(1);
+        var effect3 = effects.Single(e => e.EffectId == storedEffect3.EffectId);
+        effect3.Result.ShouldBe("some result 3".ToUtf8Bytes());
+    }
 }
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableEffectStore.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableEffectStore.cs
@@ -30,10 +30,10 @@ public Task SetEffectResult(StoredId storedId, StoredEffect storedEffect)
             ? Task.FromException(new TimeoutException())
             : _inner.SetEffectResult(storedId, storedEffect);
 
-    public Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects)
+    public Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
         => _crashed
             ? Task.FromException(new TimeoutException())
-            : _inner.SetEffectResults(storedId, storedEffects);
+            : _inner.SetEffectResults(storedId, upsertEffects, removeEffects);
 
     public Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId)
         => _crashed
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/IEffectsStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/IEffectsStore.cs
@@ -1,4 +1,5 @@
-﻿using System.Collections.Generic;
+﻿using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Cleipnir.ResilientFunctions.Storage;
@@ -8,7 +9,8 @@ public interface IEffectsStore
     Task Initialize();
     Task Truncate();
     Task SetEffectResult(StoredId storedId, StoredEffect storedEffect);
-    Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects);
+    Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects) => SetEffectResults(storedId, storedEffects, removeEffects: Array.Empty<StoredEffectId>());
+    Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects);
     Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId);
     Task DeleteEffectResult(StoredId storedId, StoredEffectId effectId);
     Task Remove(StoredId storedId);
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryEffectsStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryEffectsStore.cs
@@ -34,10 +34,13 @@ public Task SetEffectResult(StoredId storedId, StoredEffect storedEffect)
         return Task.CompletedTask;
     }
 
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects)
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
     {
-        foreach (var storedEffect in storedEffects)
+        foreach (var storedEffect in upsertEffects)
             await SetEffectResult(storedId, storedEffect);
+
+        foreach (var effectId in removeEffects)
+            await DeleteEffectResult(storedId, effectId);
     }
 
     public Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId)
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/EffectsStoreTests.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/EffectsStoreTests.cs
@@ -33,4 +33,8 @@ public override Task TruncateDeletesAllEffects()
     [TestMethod]
     public override Task BulkInsertTest()
         => BulkInsertTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
+    
+    [TestMethod]
+    public override Task BulkInsertAndDeleteTest()
+        => BulkInsertAndDeleteTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
 }
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbEffectsStore.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbEffectsStore.cs
@@ -66,7 +66,7 @@ ON DUPLICATE KEY UPDATE
     }
 
     private string? _setEffectResultsSql;
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects)
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects, IReadOnlyList<StoredEffectId> removeEffects)
     {
         await using var conn = await CreateConnection();
         _setEffectResultsSql ??= $@"
@@ -75,12 +75,19 @@ INSERT INTO {tablePrefix}_effects
           VALUES
               @VALUES  
            ON DUPLICATE KEY UPDATE
-                status = VALUES(status), result = VALUES(result), exception = VALUES(exception)";
+                status = VALUES(status), result = VALUES(result), exception = VALUES(exception);";
         
         var sql = _setEffectResultsSql.Replace(
             "@VALUES",
             "(?, ?, ?, ?, ?, ?, ?)".Replicate(storedEffects.Count).StringJoin(", ")
         );
+
+        if (removeEffects.Count > 0)
+            sql += Environment.NewLine +
+                   @$"DELETE FROM {tablePrefix}_effects 
+                      WHERE type = {storedId.Type.Value} AND 
+                            instance = '{storedId.Instance.Value:N}' AND 
+                            id_hash IN ({removeEffects.Select(id => $"'{id.Value:N}'").StringJoin(", ")});";
         
         await using var command = new MySqlCommand(sql, conn);
         foreach (var storedEffect in storedEffects)
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/EffectStoreTests.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/EffectStoreTests.cs
@@ -34,4 +34,8 @@ public override Task TruncateDeletesAllEffects()
     [TestMethod]
     public override Task BulkInsertTest()
         => BulkInsertTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
+    
+    [TestMethod]
+    public override Task BulkInsertAndDeleteTest()
+        => BulkInsertAndDeleteTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
 }
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs
@@ -1,7 +1,9 @@
 ﻿using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Helpers;
 using Cleipnir.ResilientFunctions.Storage;
 using Cleipnir.ResilientFunctions.Storage.Utils;
 using Npgsql;
@@ -68,7 +70,7 @@ ON CONFLICT (type, instance, id_hash)
         await command.ExecuteNonQueryAsync();
     }
 
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects)
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
     {
         await using var conn = await CreateConnection();
         _setEffectResultSql ??= $@"
@@ -81,7 +83,7 @@ ON CONFLICT (type, instance, id_hash)
             UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result, exception = EXCLUDED.exception";
         
         await using var batch = new NpgsqlBatch(conn);
-        foreach (var storedEffect in storedEffects)
+        foreach (var storedEffect in upsertEffects)
         {
             var command = new NpgsqlBatchCommand(_setEffectResultSql)
             {
@@ -98,6 +100,16 @@ ON CONFLICT (type, instance, id_hash)
             };
             batch.BatchCommands.Add(command);
         }
+
+        if (removeEffects.Count > 0)
+            batch.BatchCommands.Add(
+                new NpgsqlBatchCommand(
+                    @$"DELETE FROM {tablePrefix}_effects 
+                       WHERE type = {storedId.Type.Value} AND 
+                             instance = '{storedId.Instance.Value}' AND 
+                             id_hash IN ({removeEffects.Select(id => $"'{id.Value}'").StringJoin(", ")});"
+                )
+            );
         
         await batch.ExecuteNonQueryAsync();
     }
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/EffectStoreTests.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/EffectStoreTests.cs
@@ -34,4 +34,8 @@ public override Task TruncateDeletesAllEffects()
     [TestMethod]
     public override Task BulkInsertTest()
         => BulkInsertTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
+    
+    [TestMethod]
+    public override Task BulkInsertAndDeleteTest()
+        => BulkInsertAndDeleteTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));
 }
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerEffectsStore.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerEffectsStore.cs
@@ -74,8 +74,8 @@ WHEN NOT MATCHED THEN
         await command.ExecuteNonQueryAsync();
     }
 
-    private static string? _setEffectResultsSql;
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects)
+    private string? _setEffectResultsSql;
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects, IReadOnlyList<StoredEffectId> removeEffects)
     {
         await using var conn = await CreateConnection();
         _setEffectResultsSql ??= $@"
@@ -96,6 +96,14 @@ WHEN NOT MATCHED THEN
                 .Select((s, i) => s.Replace("#", i.ToString()))
                 .StringJoin(", ")
         );
+        
+        if (removeEffects.Count > 0)
+            sql += Environment.NewLine + 
+                   @$"DELETE FROM {tablePrefix}_Effects 
+                      WHERE FlowType = {storedId.Type.Value} AND 
+                            FlowInstance = '{storedId.Instance.Value}' AND 
+                            StoredId IN ({removeEffects.Select(id => $"'{id.Value}'").StringJoin(", ")}) ";
+        
         await using var command = new SqlCommand(sql, conn);
         for (var i = 0; i < storedEffects.Count; i++)
         {

Original file line number	Diff line number	Diff line change
`@@ -34,4 +34,8 @@ public override Task TruncateDeletesAllEffects()`
`34`	`34`	`[TestMethod]`
`35`	`35`	`public override Task BulkInsertTest()`
`36`	`36`	`=> BulkInsertTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));`
	`37`	`+`
	`38`	`+ [TestMethod]`
	`39`	`+ public override Task BulkInsertAndDeleteTest()`
	`40`	`+ => BulkInsertAndDeleteTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));`
`37`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -34,10 +34,13 @@ public Task SetEffectResult(StoredId storedId, StoredEffect storedEffect)`
`34`	`34`	`return Task.CompletedTask;`
`35`	`35`	`}`
`36`	`36`
`37`		`- public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> storedEffects)`
	`37`	`+ public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)`
`38`	`38`	`{`
`39`		`- foreach (var storedEffect in storedEffects)`
	`39`	`+ foreach (var storedEffect in upsertEffects)`
`40`	`40`	`await SetEffectResult(storedId, storedEffect);`
	`41`	`+`
	`42`	`+ foreach (var effectId in removeEffects)`
	`43`	`+ await DeleteEffectResult(storedId, effectId);`
`41`	`44`	`}`
`42`	`45`
`43`	`46`	`public Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId)`
Original file line number	Diff line number	Diff line change
`@@ -33,4 +33,8 @@ public override Task TruncateDeletesAllEffects()`
`33`	`33`	`[TestMethod]`
`34`	`34`	`public override Task BulkInsertTest()`
`35`	`35`	`=> BulkInsertTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));`
	`36`	`+`
	`37`	`+ [TestMethod]`
	`38`	`+ public override Task BulkInsertAndDeleteTest()`
	`39`	`+ => BulkInsertAndDeleteTest(FunctionStoreFactory.Create().SelectAsync(fs => fs.EffectsStore));`
`36`	`40`	`}`