completed rest of stores

stidsborg · stidsborg · commit 3d25d4dc3586 · 2025-06-05T12:13:57.000+02:00
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/ReplicaStoreTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/ReplicaStoreTests.cs
@@ -0,0 +1,12 @@
+﻿using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Cleipnir.ResilientFunctions.Tests.InMemoryTests;
+
+[TestClass]
+public class ReplicaStoreTests : TestTemplates.ReplicaStoreTests
+{
+    [TestMethod]
+    public override Task SunshineScenarioTest()
+        => SunshineScenarioTest(FunctionStoreFactory.Create());
+}
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs
@@ -31,7 +31,7 @@ public class InMemoryFunctionStore : IFunctionStore, IMessageStore
     
     public IMigrator Migrator { get; } = new InMemoryMigrator();
     public ISemaphoreStore SemaphoreStore { get; } = new InMemorySemaphoreStore();
-    public IReplicaStore ReplicaStore => throw new NotImplementedException(); //todo
+    public IReplicaStore ReplicaStore { get; } = new InMemoryReplicaStore();
 
     public Task Initialize() => Task.CompletedTask;
 
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryReplicaStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryReplicaStore.cs
@@ -0,0 +1,51 @@
+﻿using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Helpers;
+
+namespace Cleipnir.ResilientFunctions.Storage;
+
+public class InMemoryReplicaStore : IReplicaStore
+{
+    private readonly Dictionary<Guid, int> _replicas = new();
+    private readonly Lock _sync = new();
+    
+    public Task Initialize() => Task.CompletedTask;
+
+    public Task Insert(Guid replicaId)
+    {
+        lock (_sync)
+            _replicas.TryAdd(replicaId, 0);
+        
+        return Task.CompletedTask;
+    }
+
+    public Task Delete(Guid replicaId)
+    {
+        lock (_sync)
+            _replicas.Remove(replicaId);
+        
+        return Task.CompletedTask;
+    }
+
+    public Task UpdateHeartbeat(Guid replicaId)
+    {
+        lock (_sync)
+            if (_replicas.ContainsKey(replicaId))
+                _replicas[replicaId]++;
+        
+        return Task.CompletedTask;            
+    }
+
+    public Task<IReadOnlyList<StoredReplica>> GetAll()
+    {
+        lock (_sync)
+            return _replicas
+                .Select(kv => new StoredReplica(kv.Key, kv.Value))
+                .ToList()
+                .CastTo<IReadOnlyList<StoredReplica>>()
+                .ToTask();
+    }
+}
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/ReplicaStoreTests.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/ReplicaStoreTests.cs
@@ -0,0 +1,12 @@
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Cleipnir.ResilientFunctions.PostgreSQL.Tests;
+
+[TestClass]
+public class ReplicaStoreTests : ResilientFunctions.Tests.TestTemplates.ReplicaStoreTests
+{
+    [TestMethod]
+    public override Task SunshineScenarioTest() 
+        => SunshineScenarioTest(FunctionStoreFactory.Create());
+}
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlDbReplicaStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlDbReplicaStore.cs
@@ -0,0 +1,119 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Storage;
+using Npgsql;
+
+namespace Cleipnir.ResilientFunctions.PostgreSQL;
+
+public class PostgreSqlDbReplicaStore(string connectionString, string tablePrefix) : IReplicaStore
+{
+    private string? _initializeSql;
+    public async Task Initialize()
+    {
+        _initializeSql ??= $@"
+            CREATE TABLE IF NOT EXISTS {tablePrefix}_replicas (
+                id CHAR(32) PRIMARY KEY,
+                heartbeat INT
+            );";
+        await using var conn = await CreateConnection();
+        var command = new NpgsqlCommand(_initializeSql, conn);
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _insertSql;
+    public async Task Insert(Guid replicaId)
+    {
+        _insertSql ??= $@"
+            INSERT INTO {tablePrefix}_replicas
+                (id, heartbeat)
+            VALUES
+                ($1, 0)";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new NpgsqlCommand(_insertSql, conn)
+        {
+            Parameters =
+            {
+                new() {Value = replicaId.ToString("N")}
+            }
+        };
+
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _deleteSql;
+    public async Task Delete(Guid replicaId)
+    {
+        _deleteSql ??= $"DELETE FROM {tablePrefix}_replicas WHERE id = $1";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new NpgsqlCommand(_deleteSql, conn)
+        {
+            Parameters =
+            {
+                new() {Value = replicaId.ToString("N")}
+            }
+        };
+
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _updateHeartbeatSql;
+    public async Task UpdateHeartbeat(Guid replicaId)
+    {
+        _updateHeartbeatSql ??= $@"
+            UPDATE {tablePrefix}_replicas
+            SET heartbeat = heartbeat + 1
+            WHERE id = $1";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new NpgsqlCommand(_updateHeartbeatSql, conn)
+        {
+            Parameters =
+            {
+                new() {Value = replicaId.ToString("N")}
+            }
+        };
+
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _getAllSql;
+    public async Task<IReadOnlyList<StoredReplica>> GetAll()
+    {
+        _getAllSql ??= $"SELECT id, heartbeat FROM {tablePrefix}_replicas";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new NpgsqlCommand(_getAllSql, conn);
+        var storedReplicas = new List<StoredReplica>();
+
+        await using var reader = await command.ExecuteReaderAsync();
+        while (await reader.ReadAsync())
+        {
+            var id = Guid.Parse(reader.GetString(0));
+            var heartbeat = reader.GetInt32(1);
+            storedReplicas.Add(new StoredReplica(id, heartbeat));
+        }
+
+        return storedReplicas;
+    }
+
+    private string? _truncateSql;
+    public async Task Truncate()
+    {
+        _truncateSql ??= $"TRUNCATE TABLE {tablePrefix}_replicas";
+
+        await using var conn = await CreateConnection();
+        await using var command = new NpgsqlCommand(_truncateSql, conn);
+
+        await command.ExecuteNonQueryAsync();
+    }
+    
+    private async Task<NpgsqlConnection> CreateConnection()
+    {
+        var conn = new NpgsqlConnection(connectionString);
+        await conn.OpenAsync();
+        return conn;
+    }
+}
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs
@@ -34,7 +34,8 @@ public class PostgreSqlFunctionStore : IFunctionStore
     
     private readonly PostgreSqlSemaphoreStore _semaphoreStore;
     public ISemaphoreStore SemaphoreStore => _semaphoreStore;
-    public IReplicaStore ReplicaStore => throw new NotImplementedException();
+    private readonly PostgreSqlDbReplicaStore _replicaStore;
+    public IReplicaStore ReplicaStore => _replicaStore;
 
     public Utilities Utilities { get; }
     public IMigrator Migrator => _migrator;
@@ -57,6 +58,7 @@ public PostgreSqlFunctionStore(string connectionString, string tablePrefix = "")
         _typeStore = new PostgreSqlTypeStore(connectionString, _tableName);
         _postgresSqlUnderlyingRegister = new PostgresSqlUnderlyingRegister(connectionString, _tableName);
         _migrator = new PostgreSqlMigrator(connectionString, _tableName);
+        _replicaStore = new PostgreSqlDbReplicaStore(connectionString, _tableName);
         Utilities = new Utilities(_postgresSqlUnderlyingRegister);
     } 
 
@@ -81,6 +83,7 @@ public async Task Initialize()
         await _correlationStore.Initialize();
         await _semaphoreStore.Initialize();
         await _typeStore.Initialize();
+        await _replicaStore.Initialize();
         await using var conn = await CreateConnection();
         _initializeSql ??= $@"
             CREATE TABLE IF NOT EXISTS {_tableName} (
@@ -122,6 +125,7 @@ public async Task TruncateTables()
         await _correlationStore.Truncate();
         await _typeStore.Truncate();
         await _semaphoreStore.Truncate();
+        await _replicaStore.Truncate();
         
         await using var conn = await CreateConnection();
         _truncateTableSql ??= $"TRUNCATE TABLE {_tableName}";
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/ReplicaStoreTests.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/ReplicaStoreTests.cs
@@ -0,0 +1,12 @@
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Cleipnir.ResilientFunctions.SqlServer.Tests;
+
+[TestClass]
+public class ReplicaStoreTests : ResilientFunctions.Tests.TestTemplates.ReplicaStoreTests
+{
+    [TestMethod]
+    public override Task SunshineScenarioTest() 
+        => SunshineScenarioTest(FunctionStoreFactory.Create());
+}
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs
@@ -34,7 +34,8 @@ public class SqlServerFunctionStore : IFunctionStore
     public IMigrator Migrator => _migrator;
     private readonly SqlServerSemaphoreStore _semaphoreStore;
     public ISemaphoreStore SemaphoreStore => _semaphoreStore;
-    public IReplicaStore ReplicaStore => throw new NotImplementedException(); //todo
+    private readonly SqlServerReplicaStore _replicaStore;
+    public IReplicaStore ReplicaStore => _replicaStore;
 
     private readonly SqlServerUnderlyingRegister _underlyingRegister;
     
@@ -54,6 +55,7 @@ public SqlServerFunctionStore(string connectionString, string tablePrefix = "")
         _semaphoreStore = new SqlServerSemaphoreStore(connectionString, _tableName);
         _typeStore = new SqlServerTypeStore(connectionString, _tableName);
         _migrator = new SqlServerMigrator(connectionString, _tableName);
+        _replicaStore = new SqlServerReplicaStore(connectionString, _tableName);
         Utilities = new Utilities(_underlyingRegister);
     }
     
@@ -81,6 +83,7 @@ public async Task Initialize()
         await _correlationStore.Initialize();
         await _typeStore.Initialize();
         await _semaphoreStore.Initialize();
+        await _replicaStore.Initialize();
         await using var conn = await _connFunc();
         _initializeSql ??= @$"    
             CREATE TABLE {_tableName} (
@@ -127,6 +130,7 @@ public async Task TruncateTables()
         await _correlationStore.Truncate();
         await _typeStore.Truncate();
         await _semaphoreStore.Truncate();
+        await _replicaStore.Truncate();
         
         await using var conn = await _connFunc();
         _truncateSql ??= $"TRUNCATE TABLE {_tableName}";
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerReplicaStore.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerReplicaStore.cs
