Fix Oracle durability operation batch execution

src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs

@@ -0,0 +1,46 @@
+using Oracle.ManagedDataAccess.Client;
+using Shouldly;
+using Wolverine.Oracle;
+
+namespace OracleTests;
+
+public class oracle_durability_command_translation
+{
+    [Fact]
+    public void rewrites_generic_parameter_markers_for_oracle()
+    {
+        OracleMessageStore.normalizeParameterMarkers(
+                "delete from messages where replayable = @replayable and keep_until <= @p0")
+            .ShouldBe(
+                "delete from messages where replayable = :replayable and keep_until <= :p0");
+    }
+
+    [Fact]
+    public void separates_generic_database_batches_into_oracle_statements()
+    {
+        OracleMessageStore.splitStatements(
+                "select destination from outgoing; delete from incoming;")
+            .ShouldBe(
+            [
+                "select destination from outgoing",
+                "delete from incoming"
+            ]);
+    }
+
+    [Fact]
+    public void converts_boolean_parameters_to_oracle_number()
+    {
+        using var command = new OracleCommand();
+        command.Parameters.Add(new OracleParameter
+        {
+            ParameterName = "replayable",
+            Value = true
+        });
+
+        OracleMessageStore.normalizeParameters(command);
+
+        var parameter = (OracleParameter)command.Parameters["replayable"];
+        parameter.OracleDbType.ShouldBe(OracleDbType.Int16);
+        parameter.Value.ShouldBe(1);
+    }
+}

src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.Durability.cs

@@ -0,0 +1,133 @@
+using System.Text.RegularExpressions;
+using Oracle.ManagedDataAccess.Client;
+using Weasel.Core;
+using Wolverine.RDBMS.Polling;
+
+namespace Wolverine.Oracle;
+
+internal partial class OracleMessageStore
+{
+    async Task IDatabaseOperationBatchExecutor.ExecuteDatabaseOperationBatchAsync(
+        IDatabaseOperation[] operations,
+        CancellationToken cancellationToken)
+    {
+        await using var connection =
+            (OracleConnection)await _dataSource.OpenConnectionAsync(cancellationToken);
+        await using var transaction =
+            (OracleTransaction)await connection.BeginTransactionAsync(cancellationToken);
+
+        try
+        {
+            foreach (var operation in operations)
+            {
+                await executeOperationAsync(
+                    connection,
+                    transaction,
+                    operation,
+                    operations,
+                    cancellationToken);
+            }
+
+            await transaction.CommitAsync(cancellationToken);
+        }
+        catch
+        {
+            try
+            {
+                await transaction.RollbackAsync(cancellationToken);
+            }
+            catch
+            {
+                // Preserve the database operation failure that triggered the rollback.
+            }
+
+            throw;
+        }
+    }
+
+    private static async Task executeOperationAsync(
+        OracleConnection connection,
+        OracleTransaction transaction,
+        IDatabaseOperation operation,
+        IDatabaseOperation[] operations,
+        CancellationToken cancellationToken)
+    {
+        var builder = new DbCommandBuilder(connection);
+        operation.ConfigureCommand(builder);
+
+        await using var command = (OracleCommand)builder.Compile();
+        command.BindByName = true;
+        command.Transaction = transaction;
+
+        normalizeParameters(command);
+
+        var statements = splitStatements(command.CommandText);
+
+        if (statements.Length == 0) return;
+
+        try
+        {
+            if (operation is IDoNotReturnData)
+            {
+                foreach (var statement in statements)
+                {
+                    command.CommandText = normalizeParameterMarkers(statement);
+                    await command.ExecuteNonQueryAsync(cancellationToken);
+                }
+
+                return;
+            }
+
+            if (statements.Length != 1)
+            {
+                throw new InvalidOperationException(
+                    $"Oracle database operation '{operation}' returns data and must contain exactly one SQL statement.");
+            }
+
+            command.CommandText = normalizeParameterMarkers(statements[0]);
+            await using var reader = await command.ExecuteReaderAsync(cancellationToken);
+            var exceptions = new List<Exception>();
+            await operation.ReadResultsAsync(reader, exceptions, cancellationToken);
+            await reader.CloseAsync();
+        }
+        catch (ObjectDisposedException)
+        {
+            throw;
+        }
+        catch (Exception e)
+        {
+            throw new DatabaseBatchCommandException(command, operations, e);
+        }
+    }
+
+    internal static string normalizeParameterMarkers(string sql)
+    {
+        return Regex.Replace(sql, @"@(?=[A-Za-z_])", ":");
+    }
+
+    internal static string[] splitStatements(string sql)
+    {
+        return sql.Split(
+            ';',
+            StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
+    }
+
+    internal static void normalizeParameters(OracleCommand command)
+    {
+        foreach (OracleParameter parameter in command.Parameters)
+        {
+            switch (parameter.Value)
+            {
+                case bool boolean:
+                    parameter.OracleDbType = OracleDbType.Int16;
+                    parameter.Value = boolean ? 1 : 0;
+                    break;
+
+                case Guid guid:
+                    parameter.OracleDbType = OracleDbType.Raw;
+                    parameter.Value = guid.ToByteArray();
+                    break;
+            }
+        }
+    }
+}

src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs

@@ -36,7 +36,7 @@
 
 namespace Wolverine.Oracle;
 
-internal partial class OracleMessageStore : IMessageDatabase, IMessageInbox, IMessageOutbox, IMessageStoreAdmin, IDeadLetters, IScheduledMessages, ISagaSupport
+internal partial class OracleMessageStore : IMessageDatabase, IMessageInbox, IMessageOutbox, IMessageStoreAdmin, IDeadLetters, IScheduledMessages, ISagaSupport, IDatabaseOperationBatchExecutor
 {
     private readonly OracleDataSource _dataSource;
     private readonly DatabaseSettings _settings;

src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs

@@ -47,6 +47,21 @@ public async Task<AgentCommands> ExecuteAsync(IWolverineRuntime runtime,
     {
         if (_operations.Length == 0) return AgentCommands.Empty;
 
+        if (_database is IDatabaseOperationBatchExecutor executor)
+        {
+            try
+            {
+                await executor.ExecuteDatabaseOperationBatchAsync(_operations, cancellationToken);
+            }
+            catch (ObjectDisposedException)
+            {
+                // The system is shutting down, let this go.
+                return AgentCommands.Empty;
+            }
+
+            return postProcessingCommands();
+        }
+
         var builder = _database.ToCommandBuilder();
         foreach (var operation in _operations) operation.ConfigureCommand(builder);
 
@@ -79,12 +94,7 @@ public async Task<AgentCommands> ExecuteAsync(IWolverineRuntime runtime,
 
         try
         {
-            var commands = new AgentCommands();
-            foreach (var operation in _operations)
-            {
-                commands.AddRange(operation.PostProcessingCommands());
-            }
-            return commands;
+            return postProcessingCommands();
         }
         finally
         {
@@ -99,6 +109,17 @@ public async Task<AgentCommands> ExecuteAsync(IWolverineRuntime runtime,
         }
     }
 
+    private AgentCommands postProcessingCommands()
+    {
+        var commands = new AgentCommands();
+        foreach (var operation in _operations)
+        {
+            commands.AddRange(operation.PostProcessingCommands());
+        }
+
+        return commands;
+    }
+
     public static async Task ApplyCallbacksAsync(IReadOnlyList<IDatabaseOperation> operations, DbDataReader reader,
         IList<Exception> exceptions,
         CancellationToken token)
@@ -149,4 +170,4 @@ public static async Task ApplyCallbacksAsync(IReadOnlyList<IDatabaseOperation> o
             }
         }
     }
-}
+}

src/Persistence/Wolverine.RDBMS/Polling/IDatabaseOperation.cs

@@ -14,4 +14,11 @@ public interface IDatabaseOperation
     Task ReadResultsAsync(DbDataReader reader, IList<Exception> exceptions, CancellationToken token);
 
     IEnumerable<IAgentCommand> PostProcessingCommands();
-}
+}
+
+internal interface IDatabaseOperationBatchExecutor
+{
+    Task ExecuteDatabaseOperationBatchAsync(
+        IDatabaseOperation[] operations,
+        CancellationToken cancellationToken);
+}