Single guid id refactor for Postgres

stidsborg · stidsborg · commit d58afffba2aa · 2025-08-28T09:32:53.000+02:00
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/Utils/EnumerableLinq.cs b/Core/Cleipnir.ResilientFunctions.Tests/Utils/EnumerableLinq.cs
@@ -5,5 +5,4 @@ namespace Cleipnir.ResilientFunctions.Tests.Utils;
 public static class EnumerableLinq
 {
     public static IEnumerable<T> AsEnumerable<T>(this T t) => [t];
-    public static string JoinStrings(this IEnumerable<string> strings, string separator) => string.Join(separator, strings);
 }
diff --git a/Core/Cleipnir.ResilientFunctions/Helpers/StringExtensions.cs b/Core/Cleipnir.ResilientFunctions/Helpers/StringExtensions.cs
@@ -0,0 +1,9 @@
+using System.Collections.Generic;
+
+namespace Cleipnir.ResilientFunctions.Helpers;
+
+public static class StringExtensions
+{
+    public static string JoinStrings(this IEnumerable<string> strings, string separator)
+        => string.Join(separator, strings);
+}
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/DapperLight.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/DapperLight.cs
@@ -0,0 +1,71 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Npgsql;
+
+namespace Cleipnir.ResilientFunctions.PostgreSQL;
+
+public static class DapperLight
+{
+    public static async Task<List<T>> Query<T>(this NpgsqlConnection conn, string sql, params object[] parameters)
+    {
+        await using var cmd = new NpgsqlCommand(sql, conn);
+        foreach (var parameter in parameters)
+            cmd.Parameters.AddWithValue(parameter);
+        
+        await using var reader = await cmd.ExecuteReaderAsync();
+        var rows = new List<T>();
+        while (await reader.ReadAsync())
+            rows.Add((T) reader.GetValue(0));
+        
+        return rows;
+    }
+    
+    public static async Task<List<Row<T1, T2>>> Query<T1, T2>(this NpgsqlConnection conn, string sql, params object[] parameters)
+    {
+        await using var cmd = new NpgsqlCommand(sql, conn);
+        foreach (var parameter in parameters)
+            cmd.Parameters.AddWithValue(parameter);
+        
+        await using var reader = await cmd.ExecuteReaderAsync();
+        var rows = new List<Row<T1, T2>>();
+        while (await reader.ReadAsync())
+        {
+            var t1 = (T1) reader.GetValue(0);            
+            var t2 = (T2) reader.GetValue(1); 
+            rows.Add(new Row<T1, T2>(t1, t2));
+        }
+        
+        return rows;
+    }
+    
+    public static async Task<List<Row<T1, T2, T3>>> Query<T1, T2, T3>(this NpgsqlConnection conn, string sql, params object[] parameters)
+    {
+        await using var cmd = new NpgsqlCommand(sql, conn);
+        foreach (var parameter in parameters)
+            cmd.Parameters.AddWithValue(parameter);
+        
+        await using var reader = await cmd.ExecuteReaderAsync();
+        var rows = new List<Row<T1, T2, T3>>();
+        while (await reader.ReadAsync())
+        {
+            var t1 = (T1) reader.GetValue(0);            
+            var t2 = (T2) reader.GetValue(1); 
+            var t3 = (T3) reader.GetValue(2); 
+            rows.Add(new Row<T1, T2, T3>(t1, t2, t3));
+        }
+        
+        return rows;
+    }
+    
+    public static async Task<int> Execute(this NpgsqlConnection conn, string sql, params object[] parameters)
+    {
+        await using var cmd = new NpgsqlCommand(sql, conn);
+        foreach (var parameter in parameters)
+            cmd.Parameters.AddWithValue(parameter);
+
+        return await cmd.ExecuteNonQueryAsync();        
+    }
+}
+
+public record Row<T1, T2>(T1 Value1, T2 Value2);
+public record Row<T1, T2, T3>(T1 Value1, T2 Value2, T3 Value3);
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlCorrelationStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlCorrelationStore.cs
@@ -1,6 +1,5 @@
 ﻿using System.Collections.Generic;
 using System.Threading.Tasks;
-using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Storage;
 using Npgsql;
 
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlEffectsStore.cs
@@ -2,7 +2,6 @@
 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;
@@ -18,14 +17,13 @@ public async Task Initialize()
         await using var conn = await CreateConnection();
         _initializeSql ??= @$"
             CREATE TABLE IF NOT EXISTS {tablePrefix}_effects (
-                type INT,
-                instance UUID,
+                id UUID,
                 id_hash UUID,
                 status INT NOT NULL,
                 result BYTEA NULL,
                 exception TEXT NULL,
                 effect_id TEXT NOT NULL,
-                PRIMARY KEY (type, instance, id_hash)
+                PRIMARY KEY (id, id_hash)
             );";
         var command = new NpgsqlCommand(_initializeSql, conn);
         await command.ExecuteNonQueryAsync();
@@ -46,19 +44,18 @@ public async Task SetEffectResult(StoredId storedId, StoredEffect storedEffect)
         await using var conn = await CreateConnection();
         _setEffectResultSql ??= $@"
           INSERT INTO {tablePrefix}_effects 
-              (type, instance, id_hash, status, result, exception, effect_id)
+              (id, id_hash, status, result, exception, effect_id)
           VALUES
-              ($1, $2, $3, $4, $5, $6, $7) 
-          ON CONFLICT (type, instance, id_hash) 
+              ($1, $2, $3, $4, $5, $6) 
+          ON CONFLICT (id, id_hash) 
           DO 
             UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result, exception = EXCLUDED.exception";
         
         await using var command = new NpgsqlCommand(_setEffectResultSql, conn)
         {
             Parameters =
             {
-                new() {Value = storedId.Type.Value},
-                new() {Value = storedId.Instance.Value},
+                new() {Value = storedId.ToGuid()},
                 new() {Value = storedEffect.StoredEffectId.Value},
                 new() {Value = (int) storedEffect.WorkStatus},
                 new() {Value = storedEffect.Result ?? (object) DBNull.Value},
@@ -96,14 +93,13 @@ public async Task<IReadOnlyList<StoredEffect>> GetEffectResults(StoredId storedI
     public async Task DeleteEffectResult(StoredId storedId, StoredEffectId effectId)
     {
         await using var conn = await CreateConnection();
-        _deleteEffectResultSql ??= $"DELETE FROM {tablePrefix}_effects WHERE type = $1 AND instance = $2 AND id_hash = $3";
+        _deleteEffectResultSql ??= $"DELETE FROM {tablePrefix}_effects WHERE id = $1 AND id_hash = $2";
         
         await using var command = new NpgsqlCommand(_deleteEffectResultSql, conn)
         {
             Parameters =
             {
-                new() {Value = storedId.Type.Value },
-                new() {Value = storedId.Instance.Value },
+                new() {Value = storedId.ToGuid() },
                 new() {Value = effectId.Value },
             }
         };
@@ -116,8 +112,7 @@ public async Task DeleteEffectResults(StoredId storedId, IReadOnlyList<StoredEff
         await using var conn = await CreateConnection();
         var sql = @$"
             DELETE FROM {tablePrefix}_effects 
-            WHERE type = {storedId.Type.Value} AND 
-                  instance = '{storedId.Instance.Value}' AND 
+            WHERE id = {storedId.ToGuid()} AND                  
                   id_hash IN ({effectIds.Select(id => $"'{id.Value}'").StringJoin(", ")})";
 
         await using var command = new NpgsqlCommand(sql, conn);
@@ -128,14 +123,13 @@ DELETE FROM {tablePrefix}_effects
     public async Task Remove(StoredId storedId)
     {
         await using var conn = await CreateConnection();
-        _removeSql ??= $"DELETE FROM {tablePrefix}_effects WHERE type = $1 AND instance = $2";
+        _removeSql ??= $"DELETE FROM {tablePrefix}_effects WHERE id = $1";
         
         await using var command = new NpgsqlCommand(_removeSql, conn)
         {
             Parameters =
             {
-                new() {Value = storedId.Type.Value },
-                new() {Value = storedId.Instance.Value },
+                new() {Value = storedId.ToGuid() }
             }
         };
 
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs
@@ -87,6 +87,7 @@ public async Task Initialize()
         await using var conn = await CreateConnection();
         _initializeSql ??= $@"
             CREATE TABLE IF NOT EXISTS {_tableName} (
+                id UUID NOT NULL PRIMARY KEY,
                 type INT NOT NULL,
                 instance UUID NOT NULL,
                 epoch INT NOT NULL DEFAULT 0,
@@ -99,20 +100,19 @@ public async Task Initialize()
                 timestamp BIGINT NOT NULL,
                 human_instance_id TEXT NOT NULL,
                 parent TEXT NULL,
-                owner UUID NULL,
-                PRIMARY KEY (type, instance)
+                owner UUID NULL
             );
             CREATE INDEX IF NOT EXISTS idx_{_tableName}_expires
-            ON {_tableName}(expires, type, instance)
+            ON {_tableName}(expires, id)
             INCLUDE (epoch)
             WHERE status = {(int) Status.Executing} OR status = {(int) Status.Postponed};           
 
             CREATE INDEX IF NOT EXISTS idx_{_tableName}_succeeded
-            ON {_tableName}(type, instance)
+            ON {_tableName}(id)
             WHERE status = {(int) Status.Succeeded};
 
             CREATE INDEX IF NOT EXISTS idx_{_tableName}_owners
-            ON {_tableName}(owner, type, instance)
+            ON {_tableName}(owner)
             WHERE status = {(int) Status.Executing};
             ";
 
@@ -216,9 +216,9 @@ public async Task BulkScheduleFunctions(IEnumerable<IdWithParam> functionsWithPa
     {
         _bulkScheduleFunctionsSql ??= @$"
             INSERT INTO {_tableName}
-                (type, instance, status, param_json, expires, timestamp, human_instance_id, parent)
+                (id, type, instance, status, param_json, expires, timestamp, human_instance_id, parent)
             VALUES
-                ($1, $2, {(int) Status.Postponed}, $3, 0, 0, $4, $5)
+                ($1, $2, $3, {(int) Status.Postponed}, $4, 0, 0, $5, $6)
             ON CONFLICT DO NOTHING;";
 
         await using var conn = await CreateConnection();
@@ -232,6 +232,7 @@ INSERT INTO {_tableName}
                 {
                     Parameters =
                     {
+                        new() { Value = idWithParam.StoredId.ToGuid() },
                         new() { Value = idWithParam.StoredId.Type.Value },
                         new() { Value = idWithParam.StoredId.Instance.Value },
                         new() { Value = idWithParam.Param == null ? DBNull.Value : idWithParam.Param },
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlMessageStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlMessageStore.cs
@@ -28,13 +28,12 @@ public async Task Initialize()
         await using var conn = await CreateConnection();
         _initializeSql ??= @$"
             CREATE TABLE IF NOT EXISTS {_tablePrefix}_messages (
-                type INT,
-                instance UUID,
+                id UUID,
                 position INT NOT NULL,
                 message_json BYTEA NOT NULL,
                 message_type BYTEA NOT NULL,   
                 idempotency_key VARCHAR(255),          
-                PRIMARY KEY (type, instance, position)
+                PRIMARY KEY (id, position)
             );";
         
         var command = new NpgsqlCommand(_initializeSql, conn);
@@ -61,18 +60,17 @@ public async Task TruncateTable()
         { //append Message to message stream sql
             _appendMessageSql ??= @$"    
                 INSERT INTO {_tablePrefix}_messages
-                    (type, instance, position, message_json, message_type, idempotency_key)
+                    (id, position, message_json, message_type, idempotency_key)
                 VALUES (
-                     $1, $2, 
-                     (SELECT COALESCE(MAX(position), -1) + 1 FROM {_tablePrefix}_messages WHERE type = $1 AND instance = $2), 
-                     $3, $4, $5
+                     $1,
+                     (SELECT COALESCE(MAX(position), -1) + 1 FROM {_tablePrefix}_messages WHERE id = $1), 
+                     $2, $3, $4
                 );";
             var command = new NpgsqlBatchCommand(_appendMessageSql)
             {
                 Parameters =
                 {
-                    new() {Value = storedId.Type.Value},
-                    new() {Value = storedId.Instance.Value},
+                    new() {Value = storedId.ToGuid()},
                     new() {Value = messageJson},
                     new() {Value = messageType},
                     new() {Value = idempotencyKey ?? (object) DBNull.Value}
@@ -85,13 +83,12 @@ INSERT INTO {_tablePrefix}_messages
             _getFunctionStatusInAppendMessageSql ??= @$"    
             SELECT epoch, status
             FROM {_tablePrefix}
-            WHERE type = $1 AND instance = $2;";
+            WHERE id = $1;";
            
             var command = new NpgsqlBatchCommand(_getFunctionStatusInAppendMessageSql)
             {
                 Parameters = { 
-                    new() {Value = storedId.Type.Value},
-                    new() {Value = storedId.Instance.Value}
+                    new() {Value = storedId.ToGuid()}
                 }
             };
             batch.BatchCommands.Add(command);  
@@ -173,7 +170,7 @@ public async Task<bool> ReplaceMessage(StoredId storedId, int position, StoredMe
         _replaceMessageSql ??= @$"    
                 UPDATE {_tablePrefix}_messages
                 SET message_json = $1, message_type = $2, idempotency_key = $3
-                WHERE type = $4 AND instance = $5 AND position = $6";
+                WHERE id = $4 AND position = $5";
 
         var (messageJson, messageType, idempotencyKey) = storedMessage;
         var command = new NpgsqlCommand(_replaceMessageSql, conn)
@@ -183,8 +180,7 @@ public async Task<bool> ReplaceMessage(StoredId storedId, int position, StoredMe
                 new() {Value = messageJson},
                 new() {Value = messageType},
                 new() {Value = idempotencyKey ?? (object) DBNull.Value},
-                new() {Value = storedId.Type.Value},
-                new() {Value = storedId.Instance.Value},
+                new() {Value = storedId.ToGuid()},
                 new() {Value = position},
             }
         };
@@ -199,13 +195,12 @@ public async Task Truncate(StoredId storedId)
         await using var conn = await CreateConnection();
         _truncateFunctionSql ??= @$"    
                 DELETE FROM {_tablePrefix}_messages
-                WHERE type = $1 AND instance = $2;";
+                WHERE id = $1;";
         await using var command = new NpgsqlCommand(_truncateFunctionSql, conn)
         {
             Parameters =
             {
-                new() {Value = storedId.Type.Value},
-                new() {Value = storedId.Instance.Value}
+                new() {Value = storedId.ToGuid()}
             }
         };
         await command.ExecuteNonQueryAsync();
@@ -225,16 +220,17 @@ public async Task<IDictionary<StoredId, int>> GetMaxPositions(IReadOnlyList<Stor
     {
         if (storedIds.Count == 0)
             return new Dictionary<StoredId, int>();
-        
-        var predicates = storedIds
-            .GroupBy(id => id.Type.Value, id => id.Instance.Value)
-            .Select(g => $"type = {g.Key} AND instance IN ({g.Select(instance => $"'{instance}'").StringJoin(", ")})")
-            .StringJoin(" OR " + Environment.NewLine);
+
+        var idMap = storedIds.ToDictionary(id => id.ToGuid(), id => id);
+
+        var inClause = storedIds
+            .Select(id => $"'{id.ToGuid()}'")
+            .StringJoin(", ");
 
         var sql = @$"    
-            SELECT type, instance, position
+            SELECT id, position
             FROM {tablePrefix}_messages
-            WHERE {predicates};";
+            WHERE id IN ({inClause});";
 
         await using var conn = await CreateConnection();
         await using var command = new NpgsqlCommand(sql, conn);
@@ -246,11 +242,9 @@ public async Task<IDictionary<StoredId, int>> GetMaxPositions(IReadOnlyList<Stor
         await using var reader = await command.ExecuteReaderAsync();
         while (await reader.ReadAsync())
         {
-            var type = reader.GetInt32(0).ToStoredType();
-            var instance = reader.GetGuid(1).ToStoredInstance();
-            var storedId = new StoredId(type, instance);
-            var position = reader.GetInt32(2);
-            positions[storedId] = position;
+            var id = reader.GetGuid(0);
+            var position = reader.GetInt32(1);
+            positions[idMap[id]] = position;
         }
         
         return positions;
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/SqlGenerator.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/SqlGenerator.cs

Original file line number	Diff line number	Diff line change
`@@ -5,5 +5,4 @@ namespace Cleipnir.ResilientFunctions.Tests.Utils;`
`5`	`5`	`public static class EnumerableLinq`
`6`	`6`	`{`
`7`	`7`	`public static IEnumerable<T> AsEnumerable<T>(this T t) => [t];`
`8`		`- public static string JoinStrings(this IEnumerable<string> strings, string separator) => string.Join(separator, strings);`
`9`	`8`	`}`