Add cql batch inserts

src/SnD.Sdk.Storage.Minio/Models/EventWrapper.cs

@@ -131,7 +131,7 @@ public record S3
         /// <summary>
         /// Gets or sets the object involved in the event.
         /// </summary>
-        [JsonPropertyName("@object")]
+        [JsonPropertyName("object")]
         public Object Object { get; set; }
     }
 

src/SnD.Sdk/Logs/Providers/Configurations/DefaultLoggingConfiguration.cs

@@ -1,10 +1,12 @@
-using Serilog;
+using System.Diagnostics.CodeAnalysis;
+using Serilog;
 
 namespace Snd.Sdk.Logs.Providers.Configurations
 {
     /// <summary>
     /// Extension methods for configuration of all sinks
     /// </summary>
+    [ExcludeFromCodeCoverage]
     public static class DefaultLoggingConfiguration
     {
         /// <summary>

src/SnD.Sdk/Storage/Base/ICqlEntityService.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using Akka;
 using Akka.Streams.Dsl;
 using System.Threading.Tasks;
@@ -30,6 +31,20 @@ public interface ICqlEntityService
         /// <returns></returns>
         Task<bool> UpsertEntity<T>(T entity, int? ttlSeconds = null, bool insertNulls = false);
 
+        /// <summary>
+        /// Inserts or updates a batch of entities in the table with optional TTL and null field handling.
+        /// </summary>
+        /// <typeparam name="T">The type of the entities to be upserted.</typeparam>
+        /// <param name="entities">A list of entities to be upserted.</param>
+        /// <param name="batchSize">The number of entities to be processed in each batch. Default is 1000.</param>
+        /// <param name="ttlSeconds">Optional time to live for the entities in seconds. Default is null.</param>
+        /// <param name="insertNulls">Specifies whether to merge non-supplied fields. Default is false.</param>
+        /// <param name="rateLimit">Rate limit for the operation. Default is "1000 per second".</param>
+        /// <param name="cancellationToken">Token to monitor for cancellation requests.</param>
+        /// <returns>Returns a task that represents the asynchronous operation. The task result contains a boolean indicating success or failure.</returns>
+        Task<bool> UpsertBatch<T>(List<T> entities, int batchSize = 1000, int? ttlSeconds = null,
+            bool insertNulls = false, string rateLimit = "1000 per second", CancellationToken cancellationToken = default);
+
         /// <summary>
         /// Creates or updates a pair of entities atomically with optional TTL.
         /// </summary>

src/SnD.Sdk/Storage/Cql/CqlApiExtensions.cs

@@ -0,0 +1,83 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Polly;
+using Polly.RateLimit;
+using Polly.Retry;
+using Snd.Sdk.Tasks;
+
+namespace Snd.Sdk.Storage.Cql;
+
+/// <summary>
+/// Provides extension methods for CQL API.
+/// </summary>
+public static class CqlApiExtensions
+{
+    /// <summary>
+    /// Executes a CQL API call with retry and rate limit policies.
+    /// </summary>
+    /// <typeparam name="TResult">The type of the result produced by the CQL API call.</typeparam>
+    /// <typeparam name="TCaller">The type of the caller for logging purposes.</typeparam>
+    /// <param name="cqlApiCall">The CQL API call to be executed.</param>
+    /// <param name="logger">The logger to log retry and rate limit information.</param>
+    /// <param name="rateLimit">The rate limit string (e.g., "1000 per second").</param>
+    /// <param name="cancellationToken">The cancellation token to cancel the operation.</param>
+    /// <returns>A task that represents the asynchronous operation, which produces the result of the CQL API call.</returns>
+    public static Task<TResult> ExecuteWithRetryAndRateLimit<TResult, TCaller>(
+        this Func<CancellationToken, Task<TResult>> cqlApiCall,
+        ILogger<TCaller> logger, string rateLimit = "1000 per second",
+        CancellationToken cancellationToken = default
+    )
+    {
+        var wrapPolicy = CreateRetryPolicy(logger).WrapAsync(CreateRateLimitPolicy(rateLimit));
+        var wrappedTask = cqlApiCall.WithWrapPolicy(wrapPolicy, cancellationToken);
+
+        return wrappedTask;
+    }
+
+    private static AsyncRateLimitPolicy CreateRateLimitPolicy(string rateLimit)
+    {
+        var rateParts = rateLimit.Split(' ');
+        var limit = int.Parse(rateParts[0]);
+        var perUnit = rateParts[2];
+
+        var timeSpan = perUnit.ToLower() switch
+        {
+            "second" => TimeSpan.FromSeconds(1),
+            "minute" => TimeSpan.FromMinutes(1),
+            "hour" => TimeSpan.FromHours(1),
+            _ => throw new ArgumentException("Invalid rate limit unit.")
+        };
+
+        return Policy.RateLimitAsync(limit, timeSpan);
+    }
+
+    private static AsyncRetryPolicy CreateRetryPolicy(ILogger logger)
+    {
+        return Policy
+            .Handle<RateLimitRejectedException>()
+            .WaitAndRetryAsync(
+                retryCount: 5,
+                sleepDurationProvider: (retryAttempt, exception, context) =>
+                {
+                    // Respect the retry after time provided by the rate limiter
+                    if (exception is RateLimitRejectedException rateLimitException)
+                    {
+                        logger.LogWarning("Rate limit hit. Retrying after {RetryAfter} milliseconds",
+                            rateLimitException.RetryAfter.TotalMilliseconds);
+                        return rateLimitException.RetryAfter;
+                    }
+
+                    // Exponential backoff for other exceptions
+                    return TimeSpan.FromSeconds(Math.Pow(2, retryAttempt));
+                },
+                onRetryAsync: (exception, timeSpan, retryCount, _) =>
+                {
+                    logger.LogWarning(exception,
+                        "Retrying batch after {SleepDuration}. Retry attempt {RetryCount}",
+                        timeSpan, retryCount);
+                    return Task.CompletedTask;
+                });
+    }
+}

src/SnD.Sdk/Storage/Cql/CqlService.cs

@@ -4,7 +4,9 @@
 using Snd.Sdk.Tasks;
 using System;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Akka.Util;
 using Cassandra;
@@ -17,6 +19,7 @@ namespace Snd.Sdk.Storage.Cql
     /// <summary>
     /// CQL-API compatible entity collection service.
     /// </summary>
+    [ExcludeFromCodeCoverage]
     public class CqlService : ICqlEntityService
     {
         private readonly ILogger<CqlService> logger;
@@ -159,7 +162,50 @@ public Task<bool> UpsertEntity<T>(T entity, int? ttlSeconds = null, bool insertN
         }
 
         /// <inheritdoc />
-        public Task<bool> UpsertAtomicPair<TFirst, TSecond>(TFirst first, TSecond second, int? ttlSeconds = null, bool insertNulls = false)
+        public Task<bool> UpsertBatch<T>(List<T> entities, int batchSize = 1000, int? ttlSeconds = null,
+            bool insertNulls = false, string rateLimit = "1000 per second", CancellationToken cancellationToken = default)
+        {
+            var totalBatches = (entities.Count + batchSize - 1) / batchSize;
+            for (int i = 0; i < totalBatches; i++)
+            {
+                var batch = CreateBatch(entities, i, batchSize, ttlSeconds, insertNulls);
+                ExecuteBatch(batch, i, rateLimit, cancellationToken);
+            }
+
+            return Task.FromResult(true);
+        }
+
+        private Batch CreateBatch<T>(List<T> entities, int batchIndex, int batchSize, int? ttlSeconds, bool insertNulls)
+        {
+            var batch = this.session.CreateBatch(BatchType.Unlogged);
+            var start = batchIndex * batchSize;
+            var end = Math.Min(start + batchSize, entities.Count);
+
+            for (var j = start; j < end; j++)
+            {
+                batch.Append(GetInsertCommand(entities[j], ttlSeconds, insertNulls));
+            }
+
+            return batch;
+        }
+
+        private Task<bool> ExecuteBatch(Batch batch, int batchIndex, string rateLimit = "1000 per second", CancellationToken cancellationToken = default)
+        {
+            var cqlUpsert = (CancellationToken ct) => batch.ExecuteAsync().TryMap((() =>
+            {
+                this.logger.LogDebug("Successfully inserted batch at index {BatchIndex}. Trace: {queryTrace}", batchIndex, batch.QueryTrace);
+                return true;
+            }), exception =>
+            {
+                this.logger.LogError(exception, "Failed to insert batch at index {BatchIndex}.", batchIndex);
+                return false;
+            });
+            return cqlUpsert.ExecuteWithRetryAndRateLimit(this.logger, rateLimit, cancellationToken);
+        }
+
+        /// <inheritdoc />
+        public Task<bool> UpsertAtomicPair<TFirst, TSecond>(TFirst first, TSecond second, int? ttlSeconds = null,
+            bool insertNulls = false)
         {
             var loggedBatch = this.session.CreateBatch(BatchType.Logged);
 

src/SnD.Sdk/Storage/Providers/RedisServiceProvider.cs

@@ -1,3 +1,4 @@
+using System.Diagnostics.CodeAnalysis;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Snd.Sdk.Storage.Base;
@@ -10,6 +11,7 @@ namespace Snd.Sdk.Storage.Providers;
 /// <summary>
 /// Provider for Redis service.
 /// </summary>
+[ExcludeFromCodeCoverage]
 public static class RedisServiceProvider
 {
     /// <summary>

src/SnD.Sdk/Tasks/TaskExtensions.cs

@@ -3,6 +3,7 @@
 using System.Threading.Tasks;
 using Akka;
 using Polly.Retry;
+using Polly.Wrap;
 
 namespace Snd.Sdk.Tasks
 {
@@ -119,5 +120,18 @@ public static Task<TResult> WithRetryPolicy<TResult>(this Func<CancellationToken
         {
             return policy.ExecuteAsync(wrapped, cancellationToken);
         }
+
+        /// <summary>
+        /// Applies the specified policy wrap to the specified task and returns the result.
+        /// </summary>
+        /// <typeparam name="TResult">The type of the result produced by the task.</typeparam>
+        /// <param name="wrapped">The task to which to apply the policy wrap.</param>
+        /// <param name="policy">The policy wrap to apply.</param>
+        /// <param name="cancellationToken">Optional cancellation token for this policy wrapper.</param>
+        /// <returns>A task that represents the asynchronous operation, which produces the result of the wrapped task.</returns>
+        public static Task<TResult> WithWrapPolicy<TResult>(this Func<CancellationToken, Task<TResult>> wrapped, AsyncPolicyWrap policy, CancellationToken cancellationToken = default)
+        {
+            return policy.ExecuteAsync(wrapped, cancellationToken);
+        }
     }
 }

test/Storage/CqlTests.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Moq;
+using Snd.Sdk.Storage.Cql;
+using Xunit;
+
+namespace Snd.Sdk.Tests.Storage
+{
+    public class CqlTests : IClassFixture<AkkaFixture>, IClassFixture<LoggerFixture>
+    {
+        private readonly AkkaFixture akkaFixture;
+        private readonly LoggerFixture loggerFixture;
+
+        public CqlTests(AkkaFixture akkaFixture, LoggerFixture loggerFixture)
+        {
+            this.akkaFixture = akkaFixture;
+            this.loggerFixture = loggerFixture;
+        }
+
+        [Theory]
+        [InlineData("1000 per second", true)]
+        [InlineData("50000 per minute", true)]
+        public async Task ExecuteWithRetryAndRateLimit_ExecutesSuccessfully(string rateLimit, bool expectedResult)
+        {
+            var loggerMock = new Mock<ILogger<object>>();
+            var cqlApiCallMock = new Mock<Func<CancellationToken, Task<bool>>>();
+            cqlApiCallMock.Setup(c => c(It.IsAny<CancellationToken>())).ReturnsAsync(expectedResult);
+            var cancellationToken = CancellationToken.None;
+
+            var result = await cqlApiCallMock.Object.ExecuteWithRetryAndRateLimit(loggerMock.Object, rateLimit, cancellationToken);
+
+            Assert.True(result);
+        }
+
+        [Fact]
+        public async Task ExecuteWithRetryAndRateLimit_InvalidRateLimitUnit()
+        {
+            var loggerMock = new Mock<ILogger<object>>();
+            var cqlApiCallMock = new Mock<Func<CancellationToken, Task<int>>>();
+            var cancellationToken = CancellationToken.None;
+
+            await Assert.ThrowsAsync<ArgumentException>(() => cqlApiCallMock.Object.ExecuteWithRetryAndRateLimit(loggerMock.Object, "1000 per min", cancellationToken));
+        }
+    }
+}