Fixed effect upsert bug for postgres store

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/EffectTests.cs

@@ -77,4 +77,8 @@ public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects()
+        => UpsertingExistingEffectDoesNotAffectOtherExistingEffects(FunctionStoreFactory.Create());
 }

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/EffectTests.cs

@@ -706,5 +706,34 @@ async Task (workflow) =>
         effectIds.Single().Id.ShouldBe("SomeEffectId");
         (await cp.Effects.GetValue<Guid>("SomeEffectId")).ShouldBe(someEffectIdValue);
     }
-    
+ 
+    public abstract Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects();
+    public async Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects(Task<IFunctionStore> storeTask)
+    {  
+        var store = await storeTask;
+        var storedId = TestStoredId.Create();
+
+        var effectStore = store.EffectsStore;
+        var effectResults = new EffectResults(
+            TestFlowId.Create(),
+            storedId,
+            lazyExistingEffects: new Lazy<Task<IReadOnlyList<StoredEffect>>>(
+                () => new List<StoredEffect>().CastTo<IReadOnlyList<StoredEffect>>().ToTask()
+            ),
+            effectStore,
+            DefaultSerializer.Instance
+        );
+        var effect = new Effect(effectResults);
+
+        await effect.Capture("1", () => "hello world");
+        await effect.Capture("2", () => "hello universe");
+        await effect.Flush();
+
+        await effect.Upsert("1", "hello world again");
+        
+        var storedEffects = await effectStore.GetEffectResults(storedId);
+        storedEffects.Count.ShouldBe(2);
+        storedEffects.Single(se => se.EffectId.Id == "1").Result!.ToStringFromUtf8Bytes().DeserializeFromJsonTo<string>().ShouldBe("hello world again");
+        storedEffects.Single(se => se.EffectId.Id == "2").Result!.ToStringFromUtf8Bytes().DeserializeFromJsonTo<string>().ShouldBe("hello universe");
+    }
 }

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/RFunctionTests/EffectTests.cs

@@ -76,4 +76,8 @@ public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects()
+        => UpsertingExistingEffectDoesNotAffectOtherExistingEffects(FunctionStoreFactory.Create());
 }

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/RFunctionTests/EffectTests.cs

@@ -77,4 +77,8 @@ public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects()
+        => UpsertingExistingEffectDoesNotAffectOtherExistingEffects(FunctionStoreFactory.Create());
 }

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/SqlGenerator.cs

@@ -107,7 +107,7 @@ INSERT INTO {tablePrefix}_effects
             var sql= $@"
                 UPDATE {tablePrefix}_effects
                 SET status = $1, result = $2, exception = $3
-                WHERE type = $4 AND instance = $5;";
+                WHERE type = $4 AND instance = $5 AND id_hash = $6;";
 
            foreach (var (storedId, _, _, storedEffect) in changes.Where(s => s.Operation == CrudOperation.Update))
            {
@@ -117,6 +117,7 @@ INSERT INTO {tablePrefix}_effects
                command.AddParameter(JsonHelper.ToJson(storedEffect.StoredException) ?? (object) DBNull.Value);
                command.AddParameter(storedId.Type.Value);
                command.AddParameter(storedId.Instance.Value);
+               command.AddParameter(storedEffect.StoredEffectId.Value);
 
                commands.Add(command);
            }   

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/EffectTests.cs

@@ -77,4 +77,8 @@ public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush()
     [TestMethod]
     public override Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow()
         => CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlushInFlow(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects()
+        => UpsertingExistingEffectDoesNotAffectOtherExistingEffects(FunctionStoreFactory.Create());
 }