MySQL and SQL Server adapters, outbound delivery to external brokers, and a per-engine guide

SW.Bitween.Adapters.Bus.RabbitMq/RabbitBusHandler.cs

@@ -1,7 +1,9 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using RabbitMQ.Client;
+using Newtonsoft.Json;
 using RabbitMQ.Client.Events;
+using SW.PrimitiveTypes;
 using SW.Serverless.Sdk;
 using SW.Serverless.Sdk.Resident;
 using System;
@@ -24,8 +26,14 @@ namespace SW.Bitween.Adapters.Bus.RabbitMq;
 /// A host rejection becomes BasicNack(requeue: true), so a Bitween outage does not lose the
 /// customer's messages — it just stops draining their queue, which is the correct failure.
 /// </summary>
+// Two roles, one package. "bus" is what makes it configurable as a broker connection; "handler"
+// is what puts it in the delivery picker, because the same resident instance both consumes a
+// customer's queue and publishes back to one. Declared rather than encoded in the id: reclassifying
+// by rename would break every gateway that stores it.
 [AdapterKind("bus")]
-public class RabbitBusHandler(IOptions<RabbitOptions> options, ILogger<RabbitBusHandler> logger) : IResidentAdapter
+[AdapterKind("handler")]
+public class RabbitBusHandler(IOptions<RabbitOptions> options, ILogger<RabbitBusHandler> logger)
+    : IResidentAdapter, IInfolinkHandler
 {
     private readonly RabbitOptions _options = options.Value;
 
@@ -312,6 +320,57 @@ public Task<object> Publish(PublishRequest request)
         return Task.FromResult<object>(new { messageId, bytes = body.Length });
     }
 
+    /// <summary>
+    /// Egress through the PIPELINE: a subscription's delivery stage, publishing the message it was
+    /// given to a queue on this broker.
+    ///
+    /// <see cref="Publish"/> has existed since this adapter did, and nothing could reach it —
+    /// Bitween's pipeline calls <c>Handle</c> on a handler, and this class had none, so an operator
+    /// could consume from a customer's broker and had no way to answer on it. This is the two of
+    /// them joined up; the publishing itself is unchanged.
+    ///
+    /// Where to send is the SUBSCRIPTION's business, not the connection's: one instance serves
+    /// every gateway on this broker, so the endpoint travels with the call rather than with the
+    /// process. Deliberately NOT restricted to the endpoints the data source consumes — the common
+    /// case for egress is a queue Bitween does not drain.
+    /// </summary>
+    public Task<XchangeFile> Handle(XchangeFile xchangeFile)
+    {
+        var endpoint = _context?.ValueOf("Endpoint");
+        var exchange = _context?.ValueOf("Exchange");
+
+        if (string.IsNullOrWhiteSpace(endpoint) && string.IsNullOrWhiteSpace(exchange))
+            throw new InvalidOperationException(
+                "This delivery has no Endpoint and no Exchange, so there is nowhere to publish. Set "
+                + "Endpoint to a queue name, or Exchange (with an optional RoutingKey) to publish "
+                + "through an exchange.");
+
+        return PublishAsFile(xchangeFile, endpoint, exchange);
+    }
+
+    async Task<XchangeFile> PublishAsFile(XchangeFile xchangeFile, string endpoint, string exchange)
+    {
+        var receipt = await Publish(new PublishRequest
+        {
+            Endpoint = endpoint,
+            Exchange = exchange,
+            RoutingKey = _context?.ValueOf("RoutingKey"),
+            ContentType = _context?.ValueOf("ContentType"),
+
+            // The exchange id, so a redelivery is recognisable as the same message on the far side.
+            // Publishers that set nothing leave the consumer no way to deduplicate, which is the
+            // complaint this adapter logs when it receives one.
+            MessageId = _context?.ValueOf("xchangeid"),
+
+            Body = xchangeFile?.Data ?? ""
+        });
+
+        // The broker's receipt as the response, so what was sent and under which id is on the
+        // exchange rather than only in a log.
+        return new XchangeFile(
+            JsonConvert.SerializeObject(receipt), xchangeFile?.Filename);
+    }
+
     /// <summary>The control the UI needs before a data source is saved. Staged, so a failure names the step.</summary>
     public Task<object> TestConnection()
     {

SW.Bitween.Adapters.Bus.RabbitMq/SW.Bitween.Adapters.Bus.RabbitMq.csproj

@@ -10,6 +10,7 @@
   <ItemGroup>
     <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
     <PackageReference Include="SimplyWorks.Serverless.Sdk" Version="8.1.23" />
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.4" />
   </ItemGroup>
   <ItemGroup>
     <!-- The settings contract is source, not a package: adapters target net8.0 while the rest of

SW.Bitween.Adapters.Bus.Sqs/SqsBusHandler.cs

@@ -4,7 +4,9 @@
 using Amazon.SQS.Model;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
+using SW.PrimitiveTypes;
 using SW.Serverless.Sdk;
 using SW.Serverless.Sdk.Resident;
 using System;
@@ -37,8 +39,13 @@ namespace SW.Bitween.Adapters.Bus.Sqs;
 /// envelope they arrive in. The SP-API request/response calls themselves are ordinary HTTPS and
 /// belong in a mapper or handler, not here.
 /// </summary>
+// Two roles, one package. "bus" is what makes it configurable as a broker connection; "handler"
+// is what puts it in the delivery picker, because the same resident instance both drains a
+// customer's queue and sends back to one.
 [AdapterKind("bus")]
-public class SqsBusHandler(IOptions<SqsOptions> options, ILogger<SqsBusHandler> logger) : IResidentAdapter
+[AdapterKind("handler")]
+public class SqsBusHandler(IOptions<SqsOptions> options, ILogger<SqsBusHandler> logger)
+    : IResidentAdapter, IInfolinkHandler
 {
     private readonly SqsOptions _options = options.Value;
 
@@ -321,6 +328,47 @@ public async Task<object> Publish(PublishRequest request)
         return new { messageId = response.MessageId, sequenceNumber = response.SequenceNumber };
     }
 
+    /// <summary>
+    /// Egress through the PIPELINE: a subscription's delivery stage, sending the message it was
+    /// given to a queue these credentials can reach.
+    ///
+    /// <see cref="Publish"/> has existed since this adapter did, and nothing could reach it —
+    /// Bitween's pipeline calls <c>Handle</c> on a handler, and this class had none. This is the
+    /// two joined up; the sending itself is unchanged.
+    ///
+    /// The queue URL is the SUBSCRIPTION's, not the connection's: one instance serves every
+    /// gateway on these credentials, so it travels with the call. Deliberately not restricted to
+    /// the queues this data source consumes — the common case for egress is one it does not.
+    /// </summary>
+    public async Task<XchangeFile> Handle(XchangeFile xchangeFile)
+    {
+        var endpoint = _context?.ValueOf("Endpoint");
+
+        if (string.IsNullOrWhiteSpace(endpoint))
+            throw new InvalidOperationException(
+                "This delivery has no Endpoint, so there is nowhere to send. Set it to the queue "
+                + "URL — the full https://sqs.<region>.amazonaws.com/<account>/<name>, which is "
+                + "what the SDK addresses a queue by.");
+
+        var receipt = await Publish(new PublishRequest
+        {
+            Endpoint = endpoint,
+
+            // Only read by a FIFO queue, and required by one. Defaulting the group to the
+            // exchange id would put every message in its own group and lose the ordering a FIFO
+            // queue exists for, so it is left to configuration.
+            GroupId = _context?.ValueOf("GroupId"),
+
+            // The exchange id makes a redelivery recognisable as the same message, which is what
+            // a FIFO queue deduplicates on within its five-minute window.
+            DeduplicationId = _context?.ValueOf("DeduplicationId") ?? _context?.ValueOf("xchangeid"),
+
+            Body = xchangeFile?.Data ?? ""
+        });
+
+        return new XchangeFile(JsonConvert.SerializeObject(receipt), xchangeFile?.Filename);
+    }
+
     public async Task<object> TestConnection()
     {
         var steps = new List<object>();

SW.Bitween.Adapters.Db.Core/DbContracts.cs

@@ -70,6 +70,26 @@ public class DbCapabilities
     /// <summary>bulk, incrementing, timestamp, timestamp+incrementing, marker.</summary>
     public string[] ReceiveModes { get; set; } = Array.Empty<string>();
 
+    /// <summary>
+    /// How this engine writes a bind placeholder — <c>:</c> or <c>@</c> — filled in by the base
+    /// from the adapter's own setting rather than declared per engine, so the two cannot disagree.
+    ///
+    /// Here because a caller that WRITES SQL needs it: the schema browser drafts a statement from
+    /// a table or a procedure, and a draft using the wrong prefix is a statement the engine refuses.
+    /// </summary>
+    public string ParameterPrefix { get; set; }
+
+    /// <summary>
+    /// How this engine limits a result to the first N rows. Three shapes, and they are not
+    /// interchangeable: <c>limit</c> (PostgreSQL, MySQL) and <c>fetchFirst</c> (Oracle, and SQL
+    /// Server 2012+) go after the query, while <c>top</c> (SQL Server's idiom) goes before the
+    /// column list.
+    ///
+    /// Same reason as the prefix: a drafted statement carries a row limit, and the wrong one does
+    /// not parse.
+    /// </summary>
+    public string LimitStyle { get; set; } = "limit";
+
     /// <summary>Probed with the real credentials — what the engine allows AND this login has.</summary>
     public List<string> Privileges { get; set; } = new();
 

SW.Bitween.Adapters.Db.Core/DbResidentAdapterBase.cs

@@ -183,9 +183,46 @@ protected async Task<DbConnection> OpenAsync(CancellationToken cancellationToken
         var connection = Factory.CreateConnection();
         connection.ConnectionString = connectionString;
         await connection.OpenAsync(cancellationToken);
+        await OnConnectionOpenedAsync(connection, cancellationToken);
         return connection;
     }
 
+    /// <summary>
+    /// Narrows the declared capability list to what THIS server actually has.
+    ///
+    /// The list is written per engine and is therefore about the engine at its newest. Called once
+    /// the server version is known, which is the only point at which the difference can be told.
+    /// Does nothing unless an engine overrides it — most capabilities have been there for a decade.
+    /// </summary>
+    protected virtual void AdjustForVersion(DbCapabilities described) { }
+
+    /// <summary>
+    /// The major version this server reports, or 0 when it cannot be read. Drivers spell it
+    /// differently — "16.14", "8.4.11", "16.00.4135" — but all of them lead with the major.
+    /// </summary>
+    protected static int MajorVersionOf(string serverVersion)
+    {
+        if (string.IsNullOrWhiteSpace(serverVersion)) return 0;
+
+        var lead = serverVersion.Split('.', ' ')[0];
+        return int.TryParse(lead, out var major) ? major : 0;
+    }
+
+    /// <summary>
+    /// Runs once on every connection this adapter opens, before anything uses it.
+    ///
+    /// For the settings an engine will not take in a connection string. PostgreSQL puts search_path
+    /// there and Oracle takes CURRENT_SCHEMA the same way; SQL Server has neither, because its
+    /// default schema belongs to the login rather than to the connection — so it is the one that
+    /// needs this.
+    ///
+    /// A pooled connection carries whatever this did into its next use, which is the point: it is
+    /// paid once per physical connection rather than once per message. Anything set here therefore
+    /// has to be true for every caller of this data source, not for one of them.
+    /// </summary>
+    protected virtual Task OnConnectionOpenedAsync(DbConnection connection,
+        CancellationToken cancellationToken) => Task.CompletedTask;
+
     DbCommand CreateCommand(DbConnection connection, string sql, IDictionary<string, object> parameters,
         int? timeoutSeconds)
     {
@@ -342,6 +379,15 @@ public virtual async Task<object> Describe()
             }
         }
 
+        // Filled in here rather than declared per engine, so the prefix a caller is told to write
+        // and the prefix this adapter actually binds with cannot drift apart.
+        described.ParameterPrefix = ParameterPrefix;
+
+        // A chance for an engine to correct what it declared once it knows which VERSION it is
+        // talking to. MERGE arrived in PostgreSQL 15 and sequences in SQL Server 2012; a list that
+        // says "yes" against an older server is a statement somebody writes and cannot run.
+        AdjustForVersion(described);
+
         described.Details["statements"] = string.Join(", ", statements.Names.OrderBy(n => n));
         described.Details["allowAdHocSql"] = Options.AllowAdHocSql.ToString();
         return described;
@@ -613,17 +659,7 @@ static bool IsBareRoutineName(string sql)
 
         try
         {
-            using var command = connection.CreateCommand();
-            command.CommandText = sql;
-            command.CommandTimeout = 10;
-
-            // The placeholders have to be declared before Prepare, because some drivers validate
-            // that every parameter in the text has been supplied — Npgsql refuses outright — and
-            // a check that fell over on every parameterised statement would be worse than none.
-            DeclarePlaceholders(command);
-            PrepareCommand(command);
-            await Task.Run(() => command.Prepare());
-
+            await CheckSyntaxAsync(connection, sql);
             return (true, null);
         }
         catch (Exception ex)
@@ -632,6 +668,29 @@ static bool IsBareRoutineName(string sql)
         }
     }
 
+    /// <summary>
+    /// Asks the engine to accept this SQL without running it. Throws when it will not.
+    ///
+    /// PREPARE is the portable form and is what most drivers want: the server parses the text and
+    /// binds every name in it, which is the whole check. It is overridable because one engine
+    /// cannot be asked that way — see the SQL Server adapter, whose driver refuses to prepare a
+    /// command unless every parameter has been given an explicit type, which is precisely what a
+    /// caller checking someone else's SQL does not know.
+    /// </summary>
+    protected virtual async Task CheckSyntaxAsync(DbConnection connection, string sql)
+    {
+        using var command = connection.CreateCommand();
+        command.CommandText = sql;
+        command.CommandTimeout = 10;
+
+        // The placeholders have to be declared before Prepare, because some drivers validate that
+        // every parameter in the text has been supplied — Npgsql refuses outright — and a check
+        // that fell over on every parameterised statement would be worse than none.
+        DeclarePlaceholders(command);
+        PrepareCommand(command);
+        await Task.Run(() => command.Prepare());
+    }
+
     /// <summary>
     /// The one mistake worth naming rather than leaving to a position offset.
     ///