GH-4482 follow-up: assert the external table drains, make the store guard reachable, fix the docs

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

@@ -12,7 +12,6 @@ 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 =>
@@ -23,8 +22,11 @@ builder.UseWolverine(opts =>
     // opts.UseSqlServerPersistenceAndTransport(builder.Configuration.GetConnectionString("sqlserver"));
     // opts.UseMySqlPersistenceAndTransport(builder.Configuration.GetConnectionString("mysql"));
     // opts.UseSqlitePersistenceAndTransport(builder.Configuration.GetConnectionString("sqlite"));
-    // opts.UseOraclePersistenceAndTransport(builder.Configuration.GetConnectionString("oracle"));
-
+    // Oracle has no combined "PersistenceAndTransport" helper; its database queue transport
+    // is opted into fluently. The external table listening below needs only the persistence.
+    // opts.PersistMessagesWithOracle(builder.Configuration.GetConnectionString("oracle")!)
+    //     .EnableMessageTransport();
+
     // Or
     // opts.Services
     //     .AddMarten(builder.Configuration.GetConnectionString("postgres"))
@@ -78,8 +80,7 @@ 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>
+<sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Persistence/PostgresqlTests/Transport/external_message_tables.cs#L44-L112' 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:

src/Persistence/PersistenceTests/Transport/external_table_store_guard.cs

@@ -0,0 +1,35 @@
+using NSubstitute;
+using Xunit;
+using Shouldly;
+using Wolverine;
+using Wolverine.Persistence.Durability;
+using Wolverine.RDBMS.Transport;
+using Wolverine.Runtime;
+
+namespace PersistenceTests.Transport;
+
+public class external_table_store_guard
+{
+    /// <summary>
+    /// The guard in SendMessageThroughExternalTable used to null check the result of
+    /// JasperFx's As&lt;T&gt;(), which is a hard cast -- so the branch was unreachable and a
+    /// non-relational message store got a bare InvalidCastException instead of an explanation.
+    /// </summary>
+    [Fact]
+    public async Task a_message_store_that_cannot_do_external_tables_gets_an_explanation()
+    {
+        var runtime = Substitute.For<IWolverineRuntime>();
+        runtime.Options.Returns(new WolverineOptions());
+
+        // Deliberately NOT an IExternalDbTransportStore
+        var storage = Substitute.For<IMessageStore>();
+        runtime.Storage.Returns(storage);
+
+        var ex = await Should.ThrowAsync<InvalidOperationException>(() =>
+            runtime.SendMessageThroughExternalTable("some.table", new SomeExternalMessage()));
+
+        ex.Message.ShouldContain("relational database message storage");
+    }
+}
+
+public record SomeExternalMessage;

src/Persistence/PostgresqlTests/Transport/external_message_tables.cs

@@ -51,7 +51,10 @@ public static void Configure()
             // opts.UseSqlServerPersistenceAndTransport(builder.Configuration.GetConnectionString("sqlserver"));
             // opts.UseMySqlPersistenceAndTransport(builder.Configuration.GetConnectionString("mysql"));
             // opts.UseSqlitePersistenceAndTransport(builder.Configuration.GetConnectionString("sqlite"));
-            // opts.UseOraclePersistenceAndTransport(builder.Configuration.GetConnectionString("oracle"));
+            // Oracle has no combined "PersistenceAndTransport" helper; its database queue transport
+            // is opted into fluently. The external table listening below needs only the persistence.
+            // opts.PersistMessagesWithOracle(builder.Configuration.GetConnectionString("oracle")!)
+            //     .EnableMessageTransport();
 
             // Or
             // opts.Services

src/Persistence/SqliteTests/Transport/external_message_tables.cs

@@ -1,4 +1,6 @@
 using Microsoft.Data.Sqlite;
+using Weasel.Core;
+using Wolverine.RDBMS;
 using Weasel.Sqlite;
 using Wolverine.ComplianceTests;
 using Wolverine.Sqlite;
@@ -25,6 +27,11 @@ public external_message_tables(ITestOutputHelper output) : base(output)
     protected override void configurePersistence(WolverineOptions opts, string connectionString, string schemaName) =>
         opts.UseSqlitePersistenceAndTransport(connectionString);
 
+    // SQLite has no schemas: Wolverine folds the schema name into a table-name prefix instead
+    // (TablePrefixing, GH-3943), so "ext"."incoming1" is really the table ext_incoming1.
+    protected override string qualifiedTableNameFor(DbObjectName tableName)
+        => TablePrefixing.Apply(tableName.Schema, tableName.Name);
+
     protected override async ValueTask dropSchemaAsync(string connectionString, string[] schemas, CancellationToken cancellationToken)
     {
         await using var conn = new SqliteConnection(connectionString);

src/Persistence/Wolverine.RDBMS/Transport/ExternalDbTransportExtensions.cs

@@ -51,10 +51,14 @@ public static async Task SendMessageThroughExternalTable(this IWolverineRuntime
 
         var serializer = runtime.Options.FindSerializer(EnvelopeConstants.JsonContentType);
         var json = serializer.WriteMessage(message);
-        var database = runtime.Storage.As<IExternalDbTransportStore>();
-        if (database is null)
+        // `as`, not As<T>(): JasperFx's As<T> is a hard cast, so the null check that used to follow it
+        // here was unreachable and a non-relational message store got a bare InvalidCastException
+        // instead of this explanation. InvalidOperationException to match the equivalent guards in
+        // ExternalMessageTable.BuildListenerAsync and ExternalMessageTableListener.
+        if (runtime.Storage is not IExternalDbTransportStore database)
         {
-            throw new NotImplementedException($"The configured message store '{runtime.Storage.GetType().FullName}' does not implement {nameof(IExternalDbTransportStore)}");
+            throw new InvalidOperationException(
+                $"The external table transport option can only be used in combination with a relational database message storage option, but the message store is {runtime.Storage.GetType().FullName}");
         }
         var messageTypeName = message.GetType().ToMessageTypeName();
 

src/Testing/Wolverine.ComplianceTests/ExternalTableTransportCompliance.cs

@@ -212,9 +212,37 @@ private async Task TestMessageSend<T>(ExternalMessageTable table, T message,
 
         var envelope = tracked.Received.SingleEnvelope<T>();
         envelope.Destination.ShouldBe(new Uri($"external-table://{endpointName}/"));
+
+        // The poller deletes the source rows in the same transaction that stores them to the inbox,
+        // and only calls ReceivedAsync after that commits -- so by the time the tracked session is
+        // done the table must be empty, with no polling needed here. Asserting only on the received
+        // envelope would let a delete that quietly matches no rows pass: the message still arrives,
+        // the test still goes green, and the row is redelivered on every subsequent poll forever.
+        var storage = host.Services.GetRequiredService<IMessageStore>().As<IExternalDbTransportStore>();
+        (await countRowsAsync(storage, table.TableName, token))
+            .ShouldBe(0, $"The external message table {table.TableName.QualifiedName} should have been drained");
+
         await host.StopAsync(token);
     }
 
+    private async Task<long> countRowsAsync(IExternalDbTransportStore storage, DbObjectName tableName,
+        CancellationToken token)
+    {
+        await using var conn = await storage.DataSource.OpenConnectionAsync(token);
+        await using var cmd = conn.CreateCommand();
+        cmd.CommandText = $"select count(*) from {qualifiedTableNameFor(tableName)}";
+
+        // Convert rather than cast: providers disagree on the CLR type of count(*) -- Oracle hands
+        // back a decimal, PostgreSQL a long, SQLite a long, Sql Server an int.
+        return Convert.ToInt64(await cmd.ExecuteScalarAsync(token));
+    }
+
+    /// <summary>
+    /// How this provider spells the external table in SQL. Overridden for engines with no schemas,
+    /// where Wolverine folds the schema name into a table-name prefix instead.
+    /// </summary>
+    protected virtual string qualifiedTableNameFor(DbObjectName tableName) => tableName.QualifiedName;
+
     #endregion
 }