feat: add Postgres dev

Author: martinhonovais

src/KafkaFlow.Retry.Postgres/Repositories/IRetryQueueRepository.cs

@@ -10,7 +10,9 @@ namespace KafkaFlow.Retry.Postgres.Repositories;
 internal interface IRetryQueueRepository
 {
     Task<long> AddAsync(IDbConnection dbConnection, RetryQueueDbo retryQueueDbo);
-        
+
+    Task<long> CountQueueAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus);
+
     Task<int> DeleteQueuesAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus, DateTime maxLastExecutionDateToBeKept, int maxRowsToDelete);
 
     Task<bool> ExistsActiveAsync(IDbConnection dbConnection, string queueGroupKey);

src/KafkaFlow.Retry.Postgres/Repositories/RetryQueueRepository.cs

@@ -33,6 +33,24 @@ public async Task<long> AddAsync(IDbConnection dbConnection, RetryQueueDbo retry
         }
     }
 
+    public async Task<long> CountQueueAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus)
+    {
+        using (var command = dbConnection.CreateCommand())
+        {
+            command.CommandType = CommandType.Text;
+            command.CommandText =
+                $@"SELECT COUNT(1)
+                     FROM retry_queues
+                    WHERE SearchGroupKey = @SearchGroupKey
+                      AND IdStatus = @IdStatus";
+
+            command.Parameters.AddWithValue("SearchGroupKey", searchGroupKey);
+            command.Parameters.AddWithValue("IdStatus", (byte)retryQueueStatus);
+
+            return Convert.ToInt64(await command.ExecuteScalarAsync().ConfigureAwait(false));
+        }
+    }
+
     public async Task<int> DeleteQueuesAsync(IDbConnection dbConnection, string searchGroupKey,
         RetryQueueStatus retryQueueStatus, DateTime maxLastExecutionDateToBeKept, int maxRowsToDelete)
     {

src/KafkaFlow.Retry.Postgres/RetryQueueDataProvider.cs

@@ -107,9 +107,18 @@ public async Task<QueuePendingItemsResult> CheckQueuePendingItemsAsync(QueuePend
         }
     }
 
-    public Task<long> CountQueuesAsync(CountQueuesInput input)
+    public async Task<long> CountQueuesAsync(CountQueuesInput input)
     {
-        throw new NotImplementedException();
+        Guard.Argument(input, nameof(input)).NotNull();
+
+        using (var dbConnection = _connectionProvider.Create(_postgresDbSettings))
+        {
+            return await _retryQueueRepository.CountQueueAsync(
+                    dbConnection,
+                    input.SearchGroupKey,
+                    input.Status)
+                .ConfigureAwait(false);
+        }
     }
 
     public async Task<DeleteQueuesResult> DeleteQueuesAsync(DeleteQueuesInput input)

src/KafkaFlow.Retry.SqlServer/Repositories/IRetryQueueRepository.cs

@@ -11,14 +11,14 @@ internal interface IRetryQueueRepository
 {
     Task<long> AddAsync(IDbConnection dbConnection, RetryQueueDbo retryQueueDbo);
 
+    Task<long> CountQueueAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus);
+
     Task<int> DeleteQueuesAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus, DateTime maxLastExecutionDateToBeKept, int maxRowsToDelete);
 
     Task<bool> ExistsActiveAsync(IDbConnection dbConnection, string queueGroupKey);
 
     Task<RetryQueueDbo> GetQueueAsync(IDbConnection dbConnection, string queueGroupKey);
 
-    Task<long> CountQueueAsync(IDbConnection dbConnection, string searchGroupKey, RetryQueueStatus retryQueueStatus);
-
     Task<IList<RetryQueueDbo>> GetTopSortedQueuesOrderedAsync(IDbConnection dbConnection, RetryQueueStatus retryQueueStatus, GetQueuesSortOption sortOption, string searchGroupKey, int top);
 
     Task<int> UpdateAsync(IDbConnection dbConnection, Guid idDomain, RetryQueueStatus retryQueueStatus, DateTime lastExecution);

tests/KafkaFlow.Retry.IntegrationTests/RepositoryTests/RetryQueueDataProviderTests/CountQueuesTests.cs

@@ -18,6 +18,7 @@ public CountQueuesTests(BootstrapperRepositoryFixture bootstrapperRepositoryFixt
     [Theory]
     [InlineData(RepositoryType.MongoDb)]
     [InlineData(RepositoryType.SqlServer)]
+    [InlineData(RepositoryType.Postgres)]
     public async Task f(RepositoryType repositoryType)
     {
         // Arrange