Gh 4482 implement external table transport for oracle

docs/guide/messaging/transports/external-tables.md

@@ -2,24 +2,28 @@
 
 Let's say that you'd like to publish messages to a Wolverine application from an existing system where it's not feasible
 to either utilize Wolverine, and that system does not currently have any kind of messaging capability. And of course, you
-want the messaging to Wolverine to be robust through some sort of transactional outbox, but you certainly don't want to 
-have to build custom infrastructure to manage that. 
+want the messaging to Wolverine to be robust through some sort of transactional outbox, but you certainly don't want to
+have to build custom infrastructure to manage that.
 
 Wolverine provides a capability to scrape an externally controlled database table for incoming messages in a reliable way.
-Assuming that you are using one of the relational database options for persisting messages already like [PostgreSQL](/guide/durability/postgresql) 
-or [Sql Server](/guide/durability/sqlserver), you can tell Wolverine to poll a table *in the same database as the message 
+Assuming that you are using one of the relational database options for persisting messages already like [PostgreSQL](/guide/durability/postgresql)
+or [Sql Server](/guide/durability/sqlserver), you can tell Wolverine to poll a table *in the same database as the message
 store* for incoming messages like this:
 
 <!-- snippet: sample_configuring_external_database_messaging -->
 <a id='snippet-sample_configuring_external_database_messaging'></a>
+
 ```cs
 var builder = Host.CreateApplicationBuilder();
 builder.UseWolverine(opts =>
 {
     opts.UsePostgresqlPersistenceAndTransport(builder.Configuration.GetConnectionString("postgres")!);
 
-    // Or
+    // Or choose a different provider; MySql, Sqlite, SqlServer, and Oracle are supported.
     // opts.UseSqlServerPersistenceAndTransport(builder.Configuration.GetConnectionString("sqlserver"));
+    // opts.UseMySqlPersistenceAndTransport(builder.Configuration.GetConnectionString("mysql"));
+    // opts.UseSqlitePersistenceAndTransport(builder.Configuration.GetConnectionString("sqlite"));
+    // opts.UseOraclePersistenceAndTransport(builder.Configuration.GetConnectionString("oracle"));
 
     // Or
     // opts.Services
@@ -74,14 +78,22 @@ builder.UseWolverine(opts =>
         .Sequential();
 });
 ```
+
 <sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Persistence/PostgresqlTests/Transport/external_message_tables.cs#L233-L295' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_configuring_external_database_messaging' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
 So a couple things to know:
 
-* The external table has to have a single primary key table that uses `Guid` as the .NET type. So `uuid` for PostgreSQL or 
-  `uniqueidentifier` for Sql Server
-* There must be a single column that holds the incoming message as JSON. For Sql Server this is `varbinary(max)` and `JSONB` for PostgreSQL
+* The external table must have the following structure and datatypes. The column names can be overridden using the table configuration options.
+
+| Provider | id | body | timestamp | message_type (optional) |
+| --- | --- | --- | --- | --- |
+| PostgreSQL | UUID | JSONB | TIMESTAMP WITH TIME ZONE | VARCHAR |
+| Sql Server | UNIQUEIDENTIFIER | VARBINARY(x) | DATETIMEOFFSET | VARCHAR(x) |
+| MySQL | CHAR(x) | LONGBLOB | DATETIME | VARCHAR(x) |
+| SQLite | TEXT | TEXT | TEXT | TEXT |
+| Oracle | RAW(16) | BLOB | TIMESTAMP WITH TIME ZONE | VARCHAR2(x) |
+
 * If there is a column mapped for the message type, Wolverine is using its message type naming to determine the actual .NET
   message type. See [Message Type Name or Alias](/guide/messages.html#message-type-name-or-alias) for information about how to use this
   or even add custom type mapping to synchronize between the upstream system and your Wolverine using system

src/Persistence/MySql/MySqlTests/Transport/external_message_tables.cs

@@ -0,0 +1,34 @@
+using IntegrationTests;
+using MySqlConnector;
+using Weasel.MySql;
+using Wolverine;
+using Wolverine.ComplianceTests;
+using Wolverine.MySql;
+
+namespace MySqlTests.Transport;
+
+public class external_message_tables : ExternalTableTransportCompliance
+{
+    public external_message_tables(ITestOutputHelper output) : base(output)
+    {
+    }
+
+    protected override string connectionString => Servers.MySqlConnectionString;
+    protected override string idColumnType => "CHAR(36)";
+    protected override string bodyColumnType => "LONGBLOB";
+    protected override string timestampColumnType => "DATETIME";
+    protected override string messageTypeColumnType => "VARCHAR(255)";
+
+    protected override void configurePersistence(WolverineOptions opts, string connectionString, string schemaName) => 
+        opts.UseMySqlPersistenceAndTransport(connectionString, schemaName);
+
+    protected override async ValueTask dropSchemaAsync(string connectionString, string[] schemas, CancellationToken cancellationToken)
+    {
+        await using var conn = new MySqlConnection(connectionString);
+        await conn.OpenAsync(cancellationToken);
+        foreach (var schema in schemas)
+        {
+            await conn.DropSchemaAsync(schema, cancellationToken);
+        }
+    }
+}

src/Persistence/MySql/Wolverine.MySql/MySqlMessageStore.cs

@@ -112,11 +112,11 @@ protected override void writePagingAfter(DbCommandBuilder builder, int offset, i
         }
     }
 
-    public override ISchemaObject AddExternalMessageTable(ExternalMessageTable definition)
+    public override ITable AddExternalMessageTable(ExternalMessageTable definition)
     {
         var table = new Table(definition.TableName);
         table.AddColumn<Guid>(definition.IdColumnName).AsPrimaryKey();
-        table.AddColumn(definition.JsonBodyColumnName, "JSON").NotNull();
+        table.AddColumn(definition.JsonBodyColumnName, "LONGBLOB").NotNull();
         if (definition.TimestampColumnName.IsNotEmpty())
         {
             table.AddColumn<DateTimeOffset>(definition.TimestampColumnName)
@@ -146,8 +146,8 @@ public override async Task MigrateExternalMessageTable(ExternalMessageTable defi
         await conn.CloseAsync();
     }
 
-    protected override Task deleteMany(DbTransaction tx, Guid[] ids, DbObjectName tableName,
-        string idColumnName)
+    protected override Task deleteManyAsync(DbTransaction tx, Guid[] ids, DbObjectName tableName,
+        string idColumnName, CancellationToken token)
     {
         if (ids.Length == 0) return Task.CompletedTask;
 
@@ -157,7 +157,7 @@ protected override Task deleteMany(DbTransaction tx, Guid[] ids, DbObjectName ta
         var placeholders = MySqlCommandExtensions.WithIdList(cmd, "id", ids);
         cmd.CommandText = $"DELETE FROM {tableName.QualifiedName} WHERE {idColumnName} IN ({placeholders})";
 
-        return cmd.ExecuteNonQueryAsync();
+        return cmd.ExecuteNonQueryAsync(token);
     }
 
     protected override async Task<bool> TryAttainLockAsync(int lockId, MySqlConnection connection,
@@ -467,7 +467,7 @@ public override async Task PollForScheduledMessagesAsync(IWolverineRuntime runti
         }
     }
 
-    public override async Task PublishMessageToExternalTableAsync(ExternalMessageTable table, string messageTypeName,
+    public override async Task PublishMessageToExternalTableAsync(ExternalMessageTable table, string? messageTypeName,
         byte[] json,
         CancellationToken token)
     {
@@ -480,15 +480,15 @@ public override async Task PublishMessageToExternalTableAsync(ExternalMessageTab
             cmd.CommandText =
                 $"INSERT INTO {table.TableName.QualifiedName} ({table.IdColumnName}, {table.JsonBodyColumnName}) VALUES (@id, @json)";
             cmd.Parameters.AddWithValue("@id", Guid.NewGuid());
-            cmd.Parameters.AddWithValue("@json", json);
+            cmd.Parameters.AddWithValue("@json", System.Text.Encoding.UTF8.GetString(json));
         }
         else
         {
             cmd.CommandText =
                 $"INSERT INTO {table.TableName.QualifiedName} ({table.IdColumnName}, {table.JsonBodyColumnName}, {table.MessageTypeColumnName}) VALUES (@id, @json, @message)";
             cmd.Parameters.AddWithValue("@id", Guid.NewGuid());
-            cmd.Parameters.AddWithValue("@json", json);
-            cmd.Parameters.AddWithValue("@message", messageTypeName);
+            cmd.Parameters.AddWithValue("@json", System.Text.Encoding.UTF8.GetString(json));
+            cmd.Parameters.AddWithValue("@message", messageTypeName!);
         }
 
         await cmd.ExecuteNonQueryAsync(token);

src/Persistence/Oracle/OracleTests/Transport/external_message_tables.cs

@@ -0,0 +1,41 @@
+using System.Data.Common;
+using IntegrationTests;
+using Weasel.Core;
+using Weasel.Oracle;
+using Wolverine;
+using Wolverine.ComplianceTests;
+using Wolverine.Oracle;
+
+namespace OracleTests.Transport;
+
+public class external_message_tables : ExternalTableTransportCompliance
+{
+	public external_message_tables(ITestOutputHelper output) : base(output)
+	{
+	}
+
+	protected override string connectionString => Servers.OracleConnectionString;
+	protected override string idColumnType => "RAW(16)";
+	protected override string bodyColumnType => "BLOB";
+	protected override string timestampColumnType => "TIMESTAMP WITH TIME ZONE";
+	protected override string messageTypeColumnType => "VARCHAR2(4000)";
+
+	protected override void configurePersistence(WolverineOptions opts, string connectionString, string schemaName) =>
+		opts.PersistMessagesWithOracle(connectionString, schemaName);
+
+	protected override Task<SchemaMigration> determineMigrationAsync(DbConnection conn, CancellationToken token, ITable table)
+	{
+		var builder = new OracleMigrator().CreateCommandBuilder(conn);
+		return SchemaMigration.DetermineAsync(conn, builder, token, table);
+	}
+
+	protected override async ValueTask dropSchemaAsync(string connectionString, string[] schemas, CancellationToken cancellationToken)
+	{
+		var dataSource = new OracleDataSource(connectionString);
+		await using var conn = await dataSource.OpenConnectionAsync(cancellationToken);
+		foreach (var schema in schemas)
+		{
+			await conn.DropSchemaAsync(schema, cancellationToken);
+		}
+	}
+}

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

@@ -0,0 +1,158 @@
+using System.Data.Common;
+using JasperFx;
+using JasperFx.Core;
+using Oracle.ManagedDataAccess.Client;
+using Weasel.Core;
+using Weasel.Oracle;
+using Wolverine.Oracle.Util;
+using Wolverine.RDBMS.Transport;
+using Wolverine.Runtime;
+using Wolverine.Transports;
+using Table = Weasel.Oracle.Tables.Table;
+
+namespace Wolverine.Oracle;
+
+internal partial class OracleMessageStore
+{
+    public async Task PollForMessagesFromExternalTablesAsync(IListener listener, IWolverineRuntime settings,
+        ExternalMessageTable externalTable, IReceiver receiver, CancellationToken token)
+    {
+        // Mirrors MessageDatabase<T>.PollForMessagesFromExternalTablesAsync. The
+        // advisory lock serializes pollers across nodes; unlike the shared base
+        // (where closing the poll connection releases the lock), OracleAdvisoryLock
+        // holds its own dedicated connection, so the lock must be released explicitly
+        // after each poll or other nodes could never take over.
+        if (!await AdvisoryLock.TryAttainLockAsync(externalTable.AdvisoryLock, token))
+        {
+            return;
+        }
+
+        try
+        {
+            await using var conn = CreateConnection();
+            await conn.OpenAsync(token);
+
+            var columns = externalTable.Columns().ToArray();
+            await using var command = buildFetchSql(conn, externalTable.TableName, externalTable.Columns().ToArray(),
+                externalTable.MessageBatchSize);
+
+            var envelopes = (await command.FetchListAsync(
+                r => OracleEnvelopeReader.ReadExternalAsync(r, externalTable), token)).ToArray();
+
+            if (envelopes.Any())
+            {
+                // Important to mark every envelope as being owned by the
+                // current node
+                foreach (var envelope in envelopes)
+                {
+                    envelope.Status = EnvelopeStatus.Incoming;
+                    envelope.OwnerId = settings.DurabilitySettings.AssignedNodeNumber;
+
+                    // Fix for GH-1225
+                    envelope.Source = externalTable.TableName.QualifiedName;
+                }
+
+                var tx = await conn.BeginTransactionAsync(token);
+                await StoreIncomingAsync(tx, envelopes);
+
+                await deleteManyAsync(tx, envelopes.Select(x => x.Id).ToArray(),
+                    externalTable.TableName, externalTable.IdColumnName, token);
+                await tx.CommitAsync(token);
+
+                await receiver.ReceivedAsync(listener, envelopes);
+            }
+
+            await conn.CloseAsync();
+        }
+        catch (Exception ex)
+        {
+            await Console.Error.WriteLineAsync(ex.Message);
+        }
+        finally
+        {
+            await AdvisoryLock.ReleaseLockAsync(externalTable.AdvisoryLock);
+        }
+    }
+
+    private OracleCommand buildFetchSql(OracleConnection conn, DbObjectName tableName, string[] columnNames, int maxRecords)
+    {
+        return conn.CreateCommand($"SELECT {columnNames.Join(", ")} FROM {tableName.QualifiedName} FETCH FIRST {maxRecords} ROWS ONLY");
+    }
+
+    private static async Task deleteManyAsync(DbTransaction tx, Guid[] ids, DbObjectName tableName, string mapperIdColumnName, CancellationToken token)
+    {
+        // Oracle has no = ANY(array); bind an explicit IN list. Batch sizes are
+        // bounded by ExternalMessageTable.MessageBatchSize (default 100), well
+        // under Oracle's 1000-item IN list limit.
+        using var cmd = tx.CreateCommand("");
+        var placeholders = OracleCommandExtensions.WithIdList(cmd, "id", ids);
+        cmd.CommandText = $"DELETE FROM {tableName.QualifiedName} WHERE {mapperIdColumnName} IN ({placeholders})";
+        await cmd.ExecuteNonQueryAsync(token);
+    }
+
+    public async Task MigrateExternalMessageTable(ExternalMessageTable definition)
+    {
+        var table = AddExternalMessageTable(definition);
+        await using var conn = CreateConnection();
+        await conn.OpenAsync();
+
+        // The Oracle builder, not the default one: ODP.NET will not execute several statements from
+        // one command, and Oracle's Table registers six introspection queries as of Weasel 9.25
+        // (weasel#474). DbCommandBuilder.StartNewCommand is a no-op, so the default builder runs
+        // them together and Oracle rejects the batch with ORA-03048.
+        var builder = new OracleMigrator().CreateCommandBuilder(conn);
+        var migration = await SchemaMigration.DetermineAsync(conn, builder, CancellationToken.None, table);
+        if (migration.Difference != SchemaPatchDifference.None)
+        {
+            await new OracleMigrator().ApplyAllAsync(conn, migration, AutoCreate.CreateOrUpdate);
+        }
+
+        await conn.CloseAsync();
+    }
+
+    public async Task PublishMessageToExternalTableAsync(ExternalMessageTable table, string? messageTypeName,
+        byte[] json, CancellationToken token)
+    {
+        await using var conn = CreateConnection();
+        await conn.OpenAsync(token);
+
+        await using var cmd = conn.CreateCommand("");
+        if (table.MessageTypeColumnName.IsEmpty())
+        {
+            cmd.CommandText =
+                $"INSERT INTO {table.TableName.QualifiedName} ({table.IdColumnName}, {table.JsonBodyColumnName}) VALUES (:id, :json)";
+            cmd.With("id", Guid.NewGuid());
+            cmd.Parameters.Add(new OracleParameter("json", OracleDbType.Blob) { Value = json });
+        }
+        else
+        {
+            cmd.CommandText =
+                $"INSERT INTO {table.TableName.QualifiedName} ({table.IdColumnName}, {table.JsonBodyColumnName}, {table.MessageTypeColumnName}) VALUES (:id, :json, :message)";
+            cmd.With("id", Guid.NewGuid());
+            cmd.Parameters.Add(new OracleParameter("json", OracleDbType.Blob) { Value = json });
+            cmd.With("message", messageTypeName!);
+        }
+
+        await cmd.ExecuteNonQueryAsync(token);
+        await conn.CloseAsync();
+    }
+
+    public ITable AddExternalMessageTable(ExternalMessageTable definition)
+    {
+        var table = new Table(definition.TableName);
+        table.AddColumn<Guid>(definition.IdColumnName).AsPrimaryKey();
+        table.AddColumn(definition.JsonBodyColumnName, "BLOB").NotNull();
+        if (definition.TimestampColumnName.IsNotEmpty())
+        {
+            table.AddColumn<DateTimeOffset>(definition.TimestampColumnName)
+                .DefaultValueByExpression("SYSTIMESTAMP AT TIME ZONE ''UTC''");
+        }
+
+        if (definition.MessageTypeColumnName.IsNotEmpty())
+        {
+            table.AddColumn<string>(definition.MessageTypeColumnName);
+        }
+
+        return table;
+    }
+}