wip

Author: stidsborg

Core/Cleipnir.ResilientFunctions/Storage/Session/PositionsStorageSession.cs

@@ -0,0 +1,21 @@
+using System.Collections.Generic;
+using Cleipnir.ResilientFunctions.Domain;
+
+namespace Cleipnir.ResilientFunctions.Storage.Session;
+
+public class PositionsStorageSession : IStorageSession
+{
+    public Dictionary<string, long> Positions { get; } = new();
+    public long MaxPosition { get; set; }
+
+    public void Add(EffectId effectId)
+    {
+        var serializedEffectId = effectId.Serialize();
+        if (Positions.ContainsKey(serializedEffectId))
+            return;
+        
+        var maxPosition = MaxPosition;
+        MaxPosition++;
+        Positions[serializedEffectId] = maxPosition;
+    }
+}

Core/Cleipnir.ResilientFunctions/Storage/Types.cs

@@ -162,8 +162,9 @@ public static StoredEffect CreateState(StoredState storedState)
             StoredException: null
         );
     }
-
 };
+public record StoredEffectWithPosition(StoredEffect Effect, long Position);
+
 public record StoredState(StateId StateId, byte[] StateJson);
 
 public record IdWithParam(StoredId StoredId, string HumanInstanceId, byte[]? Param);

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs

@@ -45,7 +45,8 @@ public async Task SetEffectResults(StoredId storedId, IReadOnlyList<StoredEffect
         if (changes.Count == 0)
             return;
 
-        await using var batch = sqlGenerator.UpdateEffects(changes).ToNpgsqlBatch();
+        var positionsSession = session as PositionsStorageSession;
+        await using var batch = sqlGenerator.UpdateEffects(changes, positionsSession).ToNpgsqlBatch();
         await using var conn = await CreateConnection();
         batch.WithConnection(conn);
 
@@ -59,7 +60,7 @@ public async Task<Dictionary<StoredId, List<StoredEffect>>> GetEffectResults(IEn
 
         await using var reader = await command.ExecuteReaderAsync();
         var effects = await sqlGenerator.ReadEffectsForIds(reader, storedIds);
-        return effects;
+        return effects.ToDictionary(kv => kv.Key, kv => kv.Value.Select(s => s.Effect).ToList());
     }
 
     private string? _removeSql;
@@ -85,4 +86,29 @@ private async Task<NpgsqlConnection> CreateConnection()
         await conn.OpenAsync();
         return conn;
     }
+    
+    private async Task<PositionsStorageSession> CreateSession(StoredId storedId)
+        => await CreateSessions([storedId]).SelectAsync(d => d[storedId]);
+    private async Task<Dictionary<StoredId, PositionsStorageSession>> CreateSessions(IEnumerable<StoredId> storedIds)
+    {
+        sqlGenerator.GetEffects(storedIds);
+        
+    }
+
+    private Dictionary<StoredId, PositionsStorageSession> CreateSessions(Dictionary<StoredId, List<StoredEffectWithPosition>> effects)
+    {
+        var dictionary = new Dictionary<StoredId, PositionsStorageSession>();
+        foreach (var storedId in effects.Keys)
+        {
+            var session = new PositionsStorageSession();
+            dictionary[storedId] = session;
+            foreach (var (effect, position) in effects[storedId].OrderBy(e => e.Position))
+            {
+                session.MaxPosition = position;
+                session.Positions[effect.EffectId.Serialize()] = position;
+            }
+        }
+
+        return dictionary;
+    }
 }

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

@@ -7,6 +7,7 @@
 using Cleipnir.ResilientFunctions.Helpers;
 using Cleipnir.ResilientFunctions.Messaging;
 using Cleipnir.ResilientFunctions.Storage;
+using Cleipnir.ResilientFunctions.Storage.Session;
 using Cleipnir.ResilientFunctions.Storage.Utils;
 using Npgsql;
 
@@ -76,11 +77,11 @@ public async Task<IReadOnlyList<StoredEffect>> ReadEffects(NpgsqlDataReader read
 
         return functions;
     }
-    public async Task<Dictionary<StoredId, List<StoredEffect>>> ReadEffectsForIds(NpgsqlDataReader reader, IEnumerable<StoredId> storedIds)
+    public async Task<Dictionary<StoredId, List<StoredEffectWithPosition>>> ReadEffectsForIds(NpgsqlDataReader reader, IEnumerable<StoredId> storedIds)
     {
-        var effects = new Dictionary<StoredId, List<StoredEffect>>();
+        var effects = new Dictionary<StoredId, List<StoredEffectWithPosition>>();
         foreach (var storedId in storedIds)
-            effects[storedId] = new List<StoredEffect>();
+            effects[storedId] = new List<StoredEffectWithPosition>();
 
         while (await reader.ReadAsync())
         {
@@ -92,16 +93,16 @@ public async Task<Dictionary<StoredId, List<StoredEffect>>> ReadEffectsForIds(Np
             var effectId = reader.GetString(5);
 
             var se = new StoredEffect(EffectId.Deserialize(effectId), status, result, JsonHelper.FromJson<StoredException>(exception));
-            effects[id].Add(se);
+            effects[id].Add(new StoredEffectWithPosition(se, position));
         }
 
         return effects;
     }
 
-    public IEnumerable<StoreCommand> UpdateEffects(IReadOnlyList<StoredEffectChange> changes)
+    public IEnumerable<StoreCommand> UpdateEffects(IReadOnlyList<StoredEffectChange> changes, PositionsStorageSession session)
    {
        var commands = new List<StoreCommand>(changes.Count);
-
+        
        // INSERT
        {
            var sql= $@"
@@ -112,6 +113,7 @@ INSERT INTO {tablePrefix}_effects
 
            foreach (var (storedId, _, _, storedEffect) in changes.Where(s => s.Operation == CrudOperation.Insert))
            {
+               
                var command = StoreCommand.Create(sql);
                command.AddParameter(storedId.AsGuid);
                command.AddParameter(storedEffect!.StoredEffectId.Value.ToLong());