Feat: add Redis governance request persistence

Author: rian-be

src/Redis/Storage/Persistence/Models/RedisMutationRequestPersistenceRecord.cs

@@ -0,0 +1,18 @@
+using StackExchange.Redis;
+
+namespace ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Models;
+
+/// <summary>
+/// Represents the Redis persistence payload for governed mutation request write.
+/// </summary>
+/// <param name="RequestId">The governed request identifier.</param>
+/// <param name="DataKey">The Redis key for the serialized request document.</param>
+/// <param name="RevisionKey">The Redis key for the request revision value.</param>
+/// <param name="Payload">The serialized request payload.</param>
+/// <param name="Revision">The revision value to persist.</param>
+internal sealed record RedisMutationRequestPersistenceRecord(
+    string RequestId,
+    RedisKey DataKey,
+    RedisKey RevisionKey,
+    RedisValue Payload,
+    long Revision);

src/Redis/Storage/Persistence/Reading/RedisMutationRequestPersistenceDocumentReader.cs

@@ -0,0 +1,35 @@
+using ModularityKit.Mutator.Governance.Abstractions.Requests.Model;
+using ModularityKit.Mutator.Governance.Redis.Keys;
+using ModularityKit.Mutator.Governance.Redis.Serialization;
+using StackExchange.Redis;
+
+namespace ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Reading;
+
+/// <summary>
+/// Reads individual governed mutation requests from Redis persistence storage.
+/// </summary>
+internal sealed class RedisMutationRequestPersistenceDocumentReader(
+    IDatabase database,
+    RedisMutationRequestKeyspace keyspace)
+{
+    private readonly IDatabase _database = database ?? throw new ArgumentNullException(nameof(database));
+    private readonly RedisMutationRequestKeyspace _keyspace = keyspace ?? throw new ArgumentNullException(nameof(keyspace));
+
+    /// <summary>
+    /// Reads a single governed mutation request by identifier.
+    /// </summary>
+    /// <param name="requestId">The request identifier.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The request if it exists; otherwise <see langword="null" />.</returns>
+    public async Task<MutationRequest?> GetAsync(
+        string requestId,
+        CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        var payload = await _database.StringGetAsync(_keyspace.RequestData(requestId)).ConfigureAwait(false);
+        return payload.HasValue
+            ? RedisMutationRequestSerializer.Deserialize(payload!)
+            : null;
+    }
+}

src/Redis/Storage/Persistence/RedisMutationRequestPersistence.cs

@@ -0,0 +1,79 @@
+using ModularityKit.Mutator.Governance.Abstractions.Requests.Model;
+using ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Reading;
+using ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Writing;
+using StackExchange.Redis;
+
+namespace ModularityKit.Mutator.Governance.Redis.Storage.Persistence;
+
+/// <summary>
+/// Coordinates Redis persistence operations for governed mutation requests.
+/// </summary>
+internal sealed class RedisMutationRequestPersistence(
+    IDatabase database,
+    RedisMutationRequestPersistenceRecordFactory recordFactory,
+    RedisMutationRequestPersistenceDocumentReader documentReader,
+    RedisMutationRequestTransactionWriter transactionWriter)
+{
+    private readonly IDatabase _database = database ?? throw new ArgumentNullException(nameof(database));
+    private readonly RedisMutationRequestPersistenceRecordFactory _recordFactory = recordFactory ?? throw new ArgumentNullException(nameof(recordFactory));
+    private readonly RedisMutationRequestPersistenceDocumentReader _documentReader = documentReader ?? throw new ArgumentNullException(nameof(documentReader));
+    private readonly RedisMutationRequestTransactionWriter _transactionWriter = transactionWriter ?? throw new ArgumentNullException(nameof(transactionWriter));
+
+    /// <summary>
+    /// Creates a new governed mutation request in Redis with an initial revision.
+    /// </summary>
+    /// <param name="request">The request to create.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The persisted request with provider-managed revision values applied.</returns>
+    public async Task<MutationRequest> Create(MutationRequest request, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        cancellationToken.ThrowIfCancellationRequested();
+
+        var persistedRequest = request with { Revision = 0 };
+        var record = _recordFactory.Create(persistedRequest);
+        var transaction = _database.CreateTransaction();
+
+        _transactionWriter.WriteCreate(transaction, record, persistedRequest);
+
+        var committed = await transaction.ExecuteAsync().ConfigureAwait(false);
+        return !committed
+            ? throw new InvalidOperationException($"Mutation request '{request.RequestId}' already exists in Redis.")
+            : persistedRequest;
+    }
+
+    /// <summary>
+    /// Attempts to store an updated governed mutation request using optimistic concurrency.
+    /// </summary>
+    /// <param name="request">The request to persist.</param>
+    /// <param name="expectedRevision">The expected current revision.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The persisted request if the update succeeds; otherwise <see langword="null" />.</returns>
+    public async Task<MutationRequest?> TryStore(MutationRequest request, long expectedRevision, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        cancellationToken.ThrowIfCancellationRequested();
+
+        var currentRequest = await Get(request.RequestId, cancellationToken).ConfigureAwait(false);
+        if (currentRequest is null || currentRequest.Revision != expectedRevision)
+            return null;
+
+        var persistedRequest = request with { Revision = expectedRevision + 1 };
+        var record = _recordFactory.Create(persistedRequest);
+        var transaction = _database.CreateTransaction();
+
+        _transactionWriter.WriteUpdate(transaction, record, expectedRevision, currentRequest, persistedRequest);
+
+        var committed = await transaction.ExecuteAsync().ConfigureAwait(false);
+        return committed ? persistedRequest : null;
+    }
+
+    /// <summary>
+    /// Reads a governed mutation request by identifier.
+    /// </summary>
+    /// <param name="requestId">The request identifier.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The request if it exists; otherwise <see langword="null" />.</returns>
+    public async Task<MutationRequest?> Get(string requestId, CancellationToken cancellationToken = default) =>
+        await _documentReader.GetAsync(requestId, cancellationToken).ConfigureAwait(false);
+}

src/Redis/Storage/Persistence/Writing/RedisMutationRequestIndexWriter.cs

@@ -0,0 +1,42 @@
+using ModularityKit.Mutator.Governance.Abstractions.Requests.Model;
+using ModularityKit.Mutator.Governance.Redis.Keys;
+using StackExchange.Redis;
+
+namespace ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Writing;
+
+/// <summary>
+/// Maintains Redis secondary indexes for governed mutation request writes.
+/// </summary>
+internal sealed class RedisMutationRequestIndexWriter(
+    RedisMutationRequestKeyspace keyspace)
+{
+    private readonly RedisMutationRequestKeyspace _keyspace = keyspace ?? throw new ArgumentNullException(nameof(keyspace));
+
+    /// <summary>
+    /// Adds a request to all Redis secondary indexes implied by its current state.
+    /// </summary>
+    /// <param name="transaction">The Redis transaction to append commands to.</param>
+    /// <param name="request">The request to index.</param>
+    public void Add(ITransaction transaction, MutationRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(transaction);
+        ArgumentNullException.ThrowIfNull(request);
+
+        foreach (var indexKey in _keyspace.EnumerateIndexes(request))
+            _ = transaction.SetAddAsync(indexKey, request.RequestId);
+    }
+
+    /// <summary>
+    /// Removes a request from all Redis secondary indexes implied by its current state.
+    /// </summary>
+    /// <param name="transaction">The Redis transaction to append commands to.</param>
+    /// <param name="request">The request to remove from indexes.</param>
+    public void Remove(ITransaction transaction, MutationRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(transaction);
+        ArgumentNullException.ThrowIfNull(request);
+
+        foreach (var indexKey in _keyspace.EnumerateIndexes(request))
+            _ = transaction.SetRemoveAsync(indexKey, request.RequestId);
+    }
+}

src/Redis/Storage/Persistence/Writing/RedisMutationRequestPersistenceRecordFactory.cs

@@ -0,0 +1,32 @@
+using ModularityKit.Mutator.Governance.Abstractions.Requests.Model;
+using ModularityKit.Mutator.Governance.Redis.Keys;
+using ModularityKit.Mutator.Governance.Redis.Serialization;
+using ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Models;
+
+namespace ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Writing;
+
+/// <summary>
+/// Creates Redis persistence records from governed mutation requests.
+/// </summary>
+internal sealed class RedisMutationRequestPersistenceRecordFactory(
+    RedisMutationRequestKeyspace keyspace)
+{
+    private readonly RedisMutationRequestKeyspace _keyspace = keyspace ?? throw new ArgumentNullException(nameof(keyspace));
+
+    /// <summary>
+    /// Creates a persistence record for the supplied request.
+    /// </summary>
+    /// <param name="request">The request to convert.</param>
+    /// <returns>The Redis persistence record.</returns>
+    public RedisMutationRequestPersistenceRecord Create(MutationRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+
+        return new RedisMutationRequestPersistenceRecord(
+            request.RequestId,
+            _keyspace.RequestData(request.RequestId),
+            _keyspace.RequestRevision(request.RequestId),
+            RedisMutationRequestSerializer.Serialize(request),
+            request.Revision);
+    }
+}

src/Redis/Storage/Persistence/Writing/RedisMutationRequestTransactionWriter.cs

@@ -0,0 +1,60 @@
+using ModularityKit.Mutator.Governance.Abstractions.Requests.Model;
+using ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Models;
+using StackExchange.Redis;
+
+namespace ModularityKit.Mutator.Governance.Redis.Storage.Persistence.Writing;
+
+/// <summary>
+/// Appends Redis transaction commands for governed mutation request create and update operations.
+/// </summary>
+internal sealed class RedisMutationRequestTransactionWriter(
+    RedisMutationRequestIndexWriter indexWriter)
+{
+    private readonly RedisMutationRequestIndexWriter _indexWriter = indexWriter ?? throw new ArgumentNullException(nameof(indexWriter));
+
+    /// <summary>
+    /// Writes the Redis transaction commands required to create a new request.
+    /// </summary>
+    /// <param name="transaction">The Redis transaction to append commands to.</param>
+    /// <param name="record">The persistence record to store.</param>
+    /// <param name="request">The request being created.</param>
+    public void WriteCreate(ITransaction transaction, RedisMutationRequestPersistenceRecord record, MutationRequest request)
+    {
+        ArgumentNullException.ThrowIfNull(transaction);
+        ArgumentNullException.ThrowIfNull(record);
+        ArgumentNullException.ThrowIfNull(request);
+
+        _ = transaction.AddCondition(Condition.KeyNotExists(record.DataKey));
+        _ = transaction.AddCondition(Condition.KeyNotExists(record.RevisionKey));
+        _ = transaction.StringSetAsync(record.DataKey, record.Payload);
+        _ = transaction.StringSetAsync(record.RevisionKey, record.Revision);
+        _indexWriter.Add(transaction, request);
+    }
+
+    /// <summary>
+    /// Writes the Redis transaction commands required to update an existing request revision.
+    /// </summary>
+    /// <param name="transaction">The Redis transaction to append commands to.</param>
+    /// <param name="record">The persistence record to store.</param>
+    /// <param name="expectedRevision">The expected current revision value.</param>
+    /// <param name="currentRequest">The currently stored request state.</param>
+    /// <param name="persistedRequest">The updated request state to persist.</param>
+    public void WriteUpdate(
+        ITransaction transaction,
+        RedisMutationRequestPersistenceRecord record,
+        long expectedRevision,
+        MutationRequest currentRequest,
+        MutationRequest persistedRequest)
+    {
+        ArgumentNullException.ThrowIfNull(transaction);
+        ArgumentNullException.ThrowIfNull(record);
+        ArgumentNullException.ThrowIfNull(currentRequest);
+        ArgumentNullException.ThrowIfNull(persistedRequest);
+
+        _ = transaction.AddCondition(Condition.StringEqual(record.RevisionKey, expectedRevision));
+        _ = transaction.StringSetAsync(record.DataKey, record.Payload);
+        _ = transaction.StringSetAsync(record.RevisionKey, record.Revision);
+        _indexWriter.Remove(transaction, currentRequest);
+        _indexWriter.Add(transaction, persistedRequest);
+    }
+}