Simplified SetEffectResults-method on IEffectsStore

stidsborg · stidsborg · commit 4c24f0928f44 · 2025-03-01T08:24:04.000+01:00
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/EffectStoreTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/EffectStoreTests.cs
@@ -268,8 +268,11 @@ protected async Task BulkInsertTest(Task<IEffectsStore> storeTask)
             Result: "some result 2".ToUtf8Bytes(),
             StoredException: null
         );
-        
-        await store.SetEffectResults(storedId, [storedEffect1, storedEffect2]);
+
+        await store.SetEffectResults(
+            storedId,
+            [storedEffect1.ToStoredChange(storedId), storedEffect2.ToStoredChange(storedId)]
+        );
 
         var effects = await store.GetEffectResults(storedId);
         effects.Count.ShouldBe(2);
@@ -306,11 +309,14 @@ protected async Task BulkInsertAndDeleteTest(Task<IEffectsStore> storeTask)
             StoredException: null
         );
         
-        await store.SetEffectResults(storedId, [storedEffect1, storedEffect2]);
+        await store.SetEffectResults(storedId, [storedEffect1.ToStoredChange(storedId), storedEffect2.ToStoredChange(storedId)]);
         await store.SetEffectResults(
             storedId,
-            upsertEffects: [storedEffect3],
-            removeEffects: [storedEffect1.StoredEffectId, storedEffect2.StoredEffectId]
+            changes: [
+                storedEffect3.ToStoredChange(storedId), 
+                StoredEffectChange.CreateDelete(storedId, storedEffect1.StoredEffectId), 
+                StoredEffectChange.CreateDelete(storedId, storedEffect2.StoredEffectId)
+            ]
         );
         
         var effects = await store.GetEffectResults(storedId);
@@ -338,12 +344,17 @@ protected async Task BulkDeleteTest(Task<IEffectsStore> storeTask)
             Result: "some result 2".ToUtf8Bytes(),
             StoredException: null
         );
-        
-        await store.SetEffectResults(storedId, [storedEffect1, storedEffect2]);
+
+        await store.SetEffectResults(
+            storedId,
+            changes: [storedEffect1.ToStoredChange(storedId), storedEffect2.ToStoredChange(storedId)]
+        );
         await store.SetEffectResults(
             storedId,
-            upsertEffects: [],
-            removeEffects: [storedEffect1.StoredEffectId, storedEffect2.StoredEffectId]
+            changes: [
+                StoredEffectChange.CreateDelete(storedId, storedEffect1.StoredEffectId), 
+                StoredEffectChange.CreateDelete(storedId, storedEffect2.StoredEffectId)
+            ]
         );
         
         var effects = await store.GetEffectResults(storedId);
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> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
+    public Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes)
         => _crashed
             ? Task.FromException(new TimeoutException())
-            : _inner.SetEffectResults(storedId, upsertEffects, removeEffects);
+            : _inner.SetEffectResults(storedId, changes);
 
     public Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId)
         => _crashed
diff --git a/Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs b/Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs
@@ -358,11 +358,14 @@ private async Task Flush(StoredEffectId changedStoredId, StoredEffect? change)
                 return;
             }
 
-            await _effectsStore.SetEffectResults(
-                _storedId,
-                upsertEffects: pendingChanges.Where(pc => pc.StoredEffect != null).Select(pc => pc.StoredEffect!).ToList(),
-                removeEffects: pendingChanges.Where(pc => pc.StoredEffect == null).Select(pc => pc.Id).ToList()
-            );
+            var changes = pendingChanges
+                .Select(pc => new StoredEffectChange(
+                    _storedId,
+                    pc.Id,
+                    pc.StoredEffect == null ? CrudOperation.Delete : CrudOperation.Upsert,
+                    pc.StoredEffect)
+                ).ToList();
+            await _effectsStore.SetEffectResults(_storedId, changes);
         }
         finally
         {
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/IEffectsStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/IEffectsStore.cs
@@ -1,5 +1,4 @@
-﻿using System;
-using System.Collections.Generic;
+﻿using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Cleipnir.ResilientFunctions.Storage;
@@ -9,8 +8,7 @@ public interface IEffectsStore
     Task Initialize();
     Task Truncate();
     Task SetEffectResult(StoredId storedId, StoredEffect storedEffect);
-    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 SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes);
     Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId);
     Task DeleteEffectResult(StoredId storedId, StoredEffectId effectId);
     Task DeleteEffectResults(StoredId storedId, IReadOnlyList<StoredEffectId> effectIds);
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryEffectsStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryEffectsStore.cs
@@ -34,12 +34,12 @@ public Task SetEffectResult(StoredId storedId, StoredEffect storedEffect)
         return Task.CompletedTask;
     }
 
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes)
     {
-        foreach (var storedEffect in upsertEffects)
+        foreach (var storedEffect in changes.Where(c => c.Operation == CrudOperation.Upsert).Select(c => c.StoredEffect!))
             await SetEffectResult(storedId, storedEffect);
 
-        foreach (var effectId in removeEffects)
+        foreach (var effectId in changes.Where(c => c.Operation == CrudOperation.Delete).Select(c => c.EffectId))
             await DeleteEffectResult(storedId, effectId);
     }
 
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/Types.cs b/Core/Cleipnir.ResilientFunctions/Storage/Types.cs
@@ -74,7 +74,15 @@ public static class StoredEffectIdExtensions
     public static StoredEffectId ToStoredEffectId(this EffectId effectId) => StoredEffectId.Create(effectId);
 }
 
-public record StoredEffectChange(StoredId StoredId, StoredEffectId EffectId, CrudOperation Operation, StoredEffect? StoredEffect);
+public record StoredEffectChange(
+    StoredId StoredId,
+    StoredEffectId EffectId,
+    CrudOperation Operation,
+    StoredEffect? StoredEffect)
+{
+    public static StoredEffectChange CreateDelete(StoredId storedId, StoredEffectId effectId)
+        => new(storedId, effectId, CrudOperation.Delete, StoredEffect: null);
+}
 
 public enum CrudOperation
 {
@@ -115,4 +123,10 @@ public static StoredEffect CreateState(StoredState storedState)
 public record StoredState(StateId StateId, byte[] StateJson);
 
 public record IdWithParam(StoredId StoredId, string HumanInstanceId, byte[]? Param);
-public record LeaseUpdate(StoredId StoredId, int ExpectedEpoch);
+public record LeaseUpdate(StoredId StoredId, int ExpectedEpoch);
+
+public static class StoredEffectExtensions
+{
+    public static StoredEffectChange ToStoredChange(this StoredEffect effect, StoredId storedId) 
+        => new(storedId, effect.StoredEffectId, CrudOperation.Upsert, effect);
+}
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbEffectsStore.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbEffectsStore.cs
@@ -64,23 +64,9 @@ ON DUPLICATE KEY UPDATE
 
         await command.ExecuteNonQueryAsync();
     }
-    
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
+
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes)
     {
-        var changes = upsertEffects
-            .Select(u => new StoredEffectChange(
-                storedId,
-                u.StoredEffectId,
-                CrudOperation.Upsert,
-                u
-            ))
-            .Concat(
-                removeEffects.Select(id =>
-                    new StoredEffectChange(storedId, id, CrudOperation.Delete, StoredEffect: null)
-                )
-            )
-            .ToList();
-        
         await using var conn = await CreateConnection();
         await using var command = new MySqlCommand();
         command.Connection = conn;
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs
@@ -70,24 +70,10 @@ ON CONFLICT (type, instance, id_hash)
         await command.ExecuteNonQueryAsync();
     }
 
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes)
     {
         await using var conn = await CreateConnection();
         await using var batch = new NpgsqlBatch(conn);
-        var changes = upsertEffects
-            .Select(u => new StoredEffectChange(
-                storedId,
-                u.StoredEffectId,
-                CrudOperation.Upsert,
-                u
-            ))
-            .Concat(
-                removeEffects.Select(id =>
-                    new StoredEffectChange(storedId, id, CrudOperation.Delete, StoredEffect: null)
-                )
-            )
-            .ToList();
-        
         SqlGenerator.UpdateEffects(batch, changes, tablePrefix);        
         await batch.ExecuteNonQueryAsync();
     }
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerEffectsStore.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerEffectsStore.cs
@@ -73,30 +73,17 @@ WHEN NOT MATCHED THEN
 
         await command.ExecuteNonQueryAsync();
     }
-    
-    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)
+
+    public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes)
     {
-        var changes = upsertEffects
-            .Select(u => new StoredEffectChange(
-                storedId,
-                u.StoredEffectId,
-                CrudOperation.Upsert,
-                u
-            ))
-            .Concat(
-                removeEffects.Select(id =>
-                    new StoredEffectChange(storedId, id, CrudOperation.Delete, StoredEffect: null)
-                )
-            )
-            .ToList();
         await using var conn = await CreateConnection();
         await using var command = new SqlCommand();
         command.Connection = conn;
         command.CommandText = SqlGenerator.UpdateEffects(command, changes, tablePrefix, paramPrefix: "");
         
         await command.ExecuteNonQueryAsync();
     }
-
+    
     private string? _getEffectResultsSql;
     public async Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedId)
     {

Original file line number	Diff line number	Diff line change
`@@ -358,11 +358,14 @@ private async Task Flush(StoredEffectId changedStoredId, StoredEffect? change)`
`358`	`358`	`return;`
`359`	`359`	`}`
`360`	`360`
`361`		`- await _effectsStore.SetEffectResults(`
`362`		`- _storedId,`
`363`		`- upsertEffects: pendingChanges.Where(pc => pc.StoredEffect != null).Select(pc => pc.StoredEffect!).ToList(),`
`364`		`- removeEffects: pendingChanges.Where(pc => pc.StoredEffect == null).Select(pc => pc.Id).ToList()`
`365`		`- );`
	`361`	`+ var changes = pendingChanges`
	`362`	`+ .Select(pc => new StoredEffectChange(`
	`363`	`+ _storedId,`
	`364`	`+ pc.Id,`
	`365`	`+ pc.StoredEffect == null ? CrudOperation.Delete : CrudOperation.Upsert,`
	`366`	`+ pc.StoredEffect)`
	`367`	`+ ).ToList();`
	`368`	`+ await _effectsStore.SetEffectResults(_storedId, changes);`
`366`	`369`	`}`
`367`	`370`	`finally`
`368`	`371`	`{`
Original file line number	Diff line number	Diff line change
`@@ -34,12 +34,12 @@ 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> upsertEffects, IReadOnlyList<StoredEffectId> removeEffects)`
	`37`	`+ public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffectChange> changes)`
`38`	`38`	`{`
`39`		`- foreach (var storedEffect in upsertEffects)`
	`39`	`+ foreach (var storedEffect in changes.Where(c => c.Operation == CrudOperation.Upsert).Select(c => c.StoredEffect!))`
`40`	`40`	`await SetEffectResult(storedId, storedEffect);`
`41`	`41`
`42`		`- foreach (var effectId in removeEffects)`
	`42`	`+ foreach (var effectId in changes.Where(c => c.Operation == CrudOperation.Delete).Select(c => c.EffectId))`
`43`	`43`	`await DeleteEffectResult(storedId, effectId);`
`44`	`44`	`}`
`45`	`45`