diff --git a/SW.Bitween.Adapters.Bus.RabbitMq/RabbitBusHandler.cs b/SW.Bitween.Adapters.Bus.RabbitMq/RabbitBusHandler.cs index 335f4139..dfcdfe80 100644 --- a/SW.Bitween.Adapters.Bus.RabbitMq/RabbitBusHandler.cs +++ b/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. /// +// 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 options, ILogger logger) : IResidentAdapter +[AdapterKind("handler")] +public class RabbitBusHandler(IOptions options, ILogger logger) + : IResidentAdapter, IInfolinkHandler { private readonly RabbitOptions _options = options.Value; @@ -312,6 +320,57 @@ public Task Publish(PublishRequest request) return Task.FromResult(new { messageId, bytes = body.Length }); } + /// + /// Egress through the PIPELINE: a subscription's delivery stage, publishing the message it was + /// given to a queue on this broker. + /// + /// has existed since this adapter did, and nothing could reach it — + /// Bitween's pipeline calls Handle 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. + /// + public Task 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 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); + } + /// The control the UI needs before a data source is saved. Staged, so a failure names the step. public Task TestConnection() { diff --git a/SW.Bitween.Adapters.Bus.RabbitMq/SW.Bitween.Adapters.Bus.RabbitMq.csproj b/SW.Bitween.Adapters.Bus.RabbitMq/SW.Bitween.Adapters.Bus.RabbitMq.csproj index 78fb846f..d180c52a 100644 --- a/SW.Bitween.Adapters.Bus.RabbitMq/SW.Bitween.Adapters.Bus.RabbitMq.csproj +++ b/SW.Bitween.Adapters.Bus.RabbitMq/SW.Bitween.Adapters.Bus.RabbitMq.csproj @@ -10,6 +10,7 @@ + + + + + + + + + diff --git a/SW.Bitween.Adapters.Db.Oracle/OracleDbAdapter.cs b/SW.Bitween.Adapters.Db.Oracle/OracleDbAdapter.cs index 62015b7a..b50337a1 100644 --- a/SW.Bitween.Adapters.Db.Oracle/OracleDbAdapter.cs +++ b/SW.Bitween.Adapters.Db.Oracle/OracleDbAdapter.cs @@ -124,6 +124,78 @@ protected override void PrepareCommand(DbCommand command) // ------------------------------------------------------------------ capabilities + // ------------------------------------------------------------------ checking + + /// + /// Oracle is checked with DBMS_SQL.PARSE rather than by preparing. + /// + /// ODP.NET's Prepare is a client-side no-op — Oracle validates when a statement is + /// executed, not when it is prepared — so the shared check passed everything here, including a + /// select from a table that does not exist. A statement reported as verified without being + /// looked at is worse than no check, because it is the one nobody goes back to. + /// + /// PARSE is the real thing: it compiles the statement and resolves every name in it, raising + /// ORA-00942 for a missing table and ORA-00936 for a syntax error, and it runs nothing. Bind + /// placeholders need no values — parsing is exactly the step before binding. + /// + protected override async Task CheckSyntaxAsync(DbConnection connection, string sql) + { + using var command = connection.CreateCommand(); + command.CommandText = @" + declare + c integer := dbms_sql.open_cursor; + begin + begin + dbms_sql.parse(c, :statement, dbms_sql.native); + exception + when others then + -- Closed here as well as below: leaving a cursor open on the shared pooled + -- connection would leak one per bad statement somebody tried to save. + dbms_sql.close_cursor(c); + raise; + end; + dbms_sql.close_cursor(c); + end;"; + command.CommandTimeout = 10; + + var parameter = command.CreateParameter(); + parameter.ParameterName = "statement"; + parameter.Value = sql; + command.Parameters.Add(parameter); + + PrepareCommand(command); + + try + { + await command.ExecuteNonQueryAsync(); + } + catch (Exception ex) + { + // Rethrown without the PL/SQL stack. Running the parse inside a block means the driver + // reports where in OUR block it failed — three ORA-06512 frames through SYS.DBMS_SQL — + // underneath the one line that is about the operator's SQL. Keeping them would bury + // "table does not exist" under the mechanism used to discover it. + throw new InvalidOperationException(FirstOracleError(ex.Message), ex); + } + } + + /// + /// The first ORA- line, which is the cause; the rest are the frames it was raised through. + /// Anything that does not look like an Oracle error is returned whole rather than trimmed to + /// nothing. + /// + static string FirstOracleError(string message) + { + if (string.IsNullOrWhiteSpace(message)) return message; + + var lines = message.Split('\n', StringSplitOptions.RemoveEmptyEntries); + var cause = lines.FirstOrDefault(l => + l.TrimStart().StartsWith("ORA-", StringComparison.OrdinalIgnoreCase) + && !l.TrimStart().StartsWith("ORA-06512", StringComparison.OrdinalIgnoreCase)); + + return (cause ?? lines[0]).Trim(); + } + protected override DbCapabilities DescribeEngine() => new() { Engine = "Oracle", @@ -157,7 +229,11 @@ protected override void PrepareCommand(DbCommand command) SchemaDiscovery = true, RowCountEstimates = true, - ReceiveModes = ["bulk", "incrementing", "timestamp", "timestamp+incrementing", "marker"] + ReceiveModes = ["bulk", "incrementing", "timestamp", "timestamp+incrementing", "marker"], + + // `fetch first N rows only`, since 12c. `rownum` is the older idiom and is a trap with an + // ORDER BY — it is applied before the sort, so it takes an arbitrary N and then sorts those. + LimitStyle = "fetchFirst" }; /// diff --git a/SW.Bitween.Adapters.Db.PostgreSql/PostgreSqlDbAdapter.cs b/SW.Bitween.Adapters.Db.PostgreSql/PostgreSqlDbAdapter.cs index 48cc942d..5241d997 100644 --- a/SW.Bitween.Adapters.Db.PostgreSql/PostgreSqlDbAdapter.cs +++ b/SW.Bitween.Adapters.Db.PostgreSql/PostgreSqlDbAdapter.cs @@ -130,6 +130,25 @@ protected override string BuildConnectionString() ReceiveModes = ["bulk", "incrementing", "timestamp", "timestamp+incrementing", "marker"] }; + /// + /// MERGE arrived in PostgreSQL 15. Claiming it against 13 or 14 — both still widely run, and + /// both supported until 2025 and 2026 — would have somebody write a statement the server + /// cannot parse, and find out on the first message rather than here. + /// + protected override void AdjustForVersion(DbCapabilities described) => Narrow(described); + + /// + /// The narrowing itself, reachable without an adapter instance. Proving this needs one server + /// per version otherwise, and the rule is a comparison rather than anything the database does. + /// + public static void AdjustForVersionForTests(DbCapabilities described) => Narrow(described); + + static void Narrow(DbCapabilities described) + { + var major = MajorVersionOf(described.ServerVersion); + if (major > 0 && major < 15) described.Merge = false; + } + /// /// What this ROLE may do, asked of the server rather than assumed. Role attributes and database /// privileges are separate things in PostgreSQL and both matter — REPLICATION in particular, @@ -259,12 +278,14 @@ and n.nspname not in ('pg_catalog', 'information_schema', 'pg_toast') Type = TypeOf(reader.GetChar(2)), Comment = reader.IsDBNull(3) ? null : reader.GetString(3), - // reltuples, so it is as fresh as the last ANALYZE, and -1 on a table that has - // never been analysed. Reported as an estimate everywhere it surfaces — the - // alternative is COUNT(*) on a stranger's table, which a menu should not do. - RowCount = !request.IncludeRowCounts || reader.IsDBNull(4) + // reltuples, so it is as fresh as the last ANALYZE. It is -1 on a table that + // has never been analysed, and that is UNKNOWN rather than empty — clamping it + // to zero said "no rows" about a table that may hold millions. Reported as an + // estimate everywhere it surfaces; the alternative is COUNT(*) on a stranger's + // table, which a menu should not do. + RowCount = !request.IncludeRowCounts || reader.IsDBNull(4) || reader.GetInt64(4) < 0 ? null - : Math.Max(0, reader.GetInt64(4)) + : reader.GetInt64(4) }); } diff --git a/SW.Bitween.Adapters.Db.SqlServer/Program.cs b/SW.Bitween.Adapters.Db.SqlServer/Program.cs new file mode 100644 index 00000000..60eccd30 --- /dev/null +++ b/SW.Bitween.Adapters.Db.SqlServer/Program.cs @@ -0,0 +1,13 @@ +using Microsoft.Extensions.DependencyInjection; +using SW.Serverless.Sdk.Hosting; +using System.Threading.Tasks; + +namespace SW.Bitween.Adapters.Db.SqlServer; + +static class Program +{ + static Task Main() => AdapterHost.CreateBuilder() + .ConfigureServices((configuration, services) => services.Configure(configuration)) + .Build() + .RunResidentAsync(); +} diff --git a/SW.Bitween.Adapters.Db.SqlServer/SW.Bitween.Adapters.Db.SqlServer.csproj b/SW.Bitween.Adapters.Db.SqlServer/SW.Bitween.Adapters.Db.SqlServer.csproj new file mode 100644 index 00000000..5a1650eb --- /dev/null +++ b/SW.Bitween.Adapters.Db.SqlServer/SW.Bitween.Adapters.Db.SqlServer.csproj @@ -0,0 +1,28 @@ + + + + Exe + net8.0 + SW.Bitween.Adapters.Db.SqlServer + disable + + + + + + + + + + + + + diff --git a/SW.Bitween.Adapters.Db.SqlServer/SqlServerDbAdapter.cs b/SW.Bitween.Adapters.Db.SqlServer/SqlServerDbAdapter.cs new file mode 100644 index 00000000..a42071cc --- /dev/null +++ b/SW.Bitween.Adapters.Db.SqlServer/SqlServerDbAdapter.cs @@ -0,0 +1,619 @@ +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using SW.Serverless.Sdk; +using System; +using System.Collections.Generic; +using System.Data; +using System.Data.Common; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace SW.Bitween.Adapters.Db.SqlServer; + +/// +/// Bitween's SQL Server data source provider, which serves Azure SQL too. +/// +/// Everything generic — the command surface, paging, the statement allow-list, the polling receiver +/// — lives in . What is here is the part that is genuinely SQL +/// Server: the connection string, the sys catalog views, and what this principal is granted. +/// +// Three roles, one package. "datasource" is what makes it configurable as a connection; +// "receiver" and "handler" are what put it in the pickers a subscription actually chooses from, +// because the same resident instance both polls a table and runs a statement on delivery. +[AdapterKind("datasource")] +[AdapterKind("receiver")] +[AdapterKind("handler")] +public class SqlServerDbAdapter(IOptions options, ILogger logger) + : DbResidentAdapterBase(options.Value, logger) +{ + readonly SqlServerOptions _options = options.Value; + + protected override DbProviderFactory Factory => SqlClientFactory.Instance; + + protected override string ParameterPrefix => "@"; + + // ------------------------------------------------------------------ connection + + protected override string BuildConnectionString() + { + if (string.IsNullOrWhiteSpace(_options.Database)) + throw new InvalidOperationException( + "Database is required: this connects to one database on the server, and it is what " + + "an unqualified name in a statement resolves against."); + + // A named instance is resolved through the SQL Server Browser, not by port, and supplying + // both is an error the driver reports obscurely — so the port is only added when the host + // is not already naming an instance. + var host = (_options.Host ?? "").Trim(); + var dataSource = host.Contains('\\') || host.Contains(',') + ? host + : $"{host},{Math.Max(1, _options.Port)}"; + + var builder = new SqlConnectionStringBuilder + { + DataSource = dataSource, + InitialCatalog = _options.Database, + UserID = _options.UserName, + Password = _options.Password ?? "", + ConnectTimeout = Math.Max(1, _options.ConnectTimeoutSeconds), + CommandTimeout = Math.Max(0, _options.CommandTimeoutSeconds), + ApplicationName = _options.ApplicationName ?? "Bitween", + + // The whole reason this adapter is resident: the pool outlives the message, so a + // connect, a TLS handshake and an authentication round trip are paid once rather than + // per Xchange. + Pooling = true, + MinPoolSize = Math.Max(0, _options.MinPoolSize), + MaxPoolSize = Math.Max(1, _options.MaxPoolSize), + + Encrypt = _options.Encrypt, + TrustServerCertificate = _options.TrustServerCertificate, + MultipleActiveResultSets = _options.MultipleActiveResultSets + }; + + return builder.ToString(); + } + + /// + /// SQL Server has no search_path: the default schema belongs to the LOGIN, not to the + /// connection. Where a data source names one, it is applied per connection instead. + /// + /// Failure is deliberately not fatal. Impersonating a schema's owner is a privilege many + /// service accounts do not have, and a connection that works perfectly for qualified names + /// should not be refused because it could not take a shortcut for unqualified ones. + /// + protected override async Task OnConnectionOpenedAsync(DbConnection connection, + CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(_options.Schema) && !_options.SnapshotIsolation) return; + + try + { + if (_options.SnapshotIsolation) + { + using var isolation = connection.CreateCommand(); + isolation.CommandText = "set transaction isolation level snapshot"; + isolation.CommandTimeout = 10; + await isolation.ExecuteNonQueryAsync(cancellationToken); + } + + if (!string.IsNullOrWhiteSpace(_options.Schema)) + { + using var command = connection.CreateCommand(); + + // The owner of the schema, which is the principal whose default schema it is. Bound + // as a parameter and then emitted through QUOTENAME, because EXECUTE AS takes a + // literal and this value comes from configuration. + command.CommandText = @" + declare @owner sysname = + (select dp.name + from sys.schemas s + join sys.database_principals dp on dp.principal_id = s.principal_id + where s.name = @schema); + if @owner is not null and @owner <> user_name() + exec('execute as user = ' + quotename(@owner));"; + command.CommandTimeout = 10; + + var parameter = command.CreateParameter(); + parameter.ParameterName = "schema"; + parameter.Value = _options.Schema; + command.Parameters.Add(parameter); + + await command.ExecuteNonQueryAsync(cancellationToken); + } + } + catch (Exception ex) + { + Logger.LogDebug(ex, + "Could not apply the connection defaults for schema {Schema}; unqualified names " + + "will resolve against the login's own default schema.", _options.Schema); + } + } + + // ------------------------------------------------------------------ checking + + /// + /// SQL Server is checked with sp_describe_undeclared_parameters rather than by preparing. + /// + /// SqlCommand.Prepare refuses unless every parameter has been given an explicit type and + /// size — which is exactly what a caller checking somebody else's SQL does not know, and + /// guessing nvarchar(4000) for a parameter compared against an int would make the check answer + /// a different question from the one asked. + /// + /// The procedure parses the batch and binds every name in it, then reports the parameters it + /// would need — so it catches a syntax error and a missing table, which are the two things this + /// check exists to catch, and needs nothing supplied. + /// + protected override async Task CheckSyntaxAsync(DbConnection connection, string sql) + { + using var command = connection.CreateCommand(); + command.CommandText = "sys.sp_describe_undeclared_parameters"; + command.CommandType = CommandType.StoredProcedure; + command.CommandTimeout = 10; + + var parameter = command.CreateParameter(); + parameter.ParameterName = "@tsql"; + parameter.DbType = DbType.String; + parameter.Size = -1; + parameter.Value = sql; + command.Parameters.Add(parameter); + + // ExecuteReader rather than ExecuteNonQuery: the answer is a result set, and a batch that + // will not bind throws while producing it. + using var reader = await command.ExecuteReaderAsync(); + while (await reader.ReadAsync()) { } + } + + // ------------------------------------------------------------------ capabilities + + protected override DbCapabilities DescribeEngine() => new() + { + Engine = "SQL Server", + SupportedObjects = ["table", "view", "procedure", "function", "sequence"], + + StoredProcedures = true, + ProcedureOutParameters = true, + + // A procedure returns result sets by SELECTing, with nothing to declare and nothing to + // bind — so Call returns rows directly, where PostgreSQL needs a set-returning function + // queried with SELECT and Oracle needs an explicit REF CURSOR. + ProcedureResultSets = true, + MultipleResultSets = true, + + NamedParameters = true, + Transactions = true, + + // Snapshot as well as the four standard levels, and it is the one worth having here: a + // long-running read for a polling receiver does not block the writers it is reading from. + IsolationLevels = + ["ReadUncommitted", "ReadCommitted", "RepeatableRead", "Serializable", "Snapshot"], + + // SqlBulkCopy exists and is the fastest path there is, but BulkLoad is not implemented for + // it yet — false rather than advertising a path that would quietly fall back to row-by-row. + BulkCopy = false, + + // The real thing, not an approximation of one. + Merge = true, + + // The OUTPUT clause, which does what RETURNING does and does it for MERGE too. + Returning = true, + + Json = true, + + // Table-valued parameters are the nearest thing and are not an array type; a caller cannot + // bind a list to one parameter the way PostgreSQL allows. + ArrayTypes = false, + + // Query Notifications over Service Broker is exactly the thing a resident adapter could + // hold for free, and it is not wired yet. Declared false so the UI says "not available" + // rather than leaving a gap. Change Tracking and CDC are the same story. + ChangeNotification = false, + LogBasedCdc = false, + + SchemaDiscovery = true, + RowCountEstimates = true, + ReceiveModes = ["bulk", "incrementing", "timestamp", "timestamp+incrementing", "marker"], + + // TOP, which goes before the column list rather than after the query. OFFSET/FETCH exists + // too and is what paging uses, but it requires an ORDER BY — TOP is the form that works on + // a draft nobody has ordered yet. + LimitStyle = "top" + }; + + /// + /// What this principal may do IN THIS DATABASE, asked of the server rather than assumed. + /// fn_my_permissions accounts for role membership and for permissions granted at the server + /// level that reach down, which reading sys.database_permissions directly does not. + /// + protected override async Task> ProbePrivilegesAsync(DbConnection connection, + CancellationToken cancellationToken) + { + var privileges = new List(); + + try + { + using var command = connection.CreateCommand(); + command.CommandText = + "select permission_name from fn_my_permissions(null, 'DATABASE') order by permission_name"; + command.CommandTimeout = 10; + + using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + privileges.Add(reader.GetString(0)); + } + catch (Exception ex) + { + // A principal that cannot call fn_my_permissions is unusual but not broken — it just + // cannot tell us what it can do. Saying so beats failing the connection test over it. + Logger.LogDebug(ex, "Could not read permissions."); + privileges.Add($"(could not be read: {ex.Message})"); + } + + return privileges; + } + + protected override IEnumerable> ExtraStatusDetails() + { + yield return new KeyValuePair("sqlserver.database", _options.Database ?? ""); + yield return new KeyValuePair("sqlserver.schema", _options.Schema ?? "(login default)"); + yield return new KeyValuePair("sqlserver.encrypt", _options.Encrypt.ToString()); + yield return new KeyValuePair( + "sqlserver.trustServerCertificate", _options.TrustServerCertificate.ToString()); + } + + // ------------------------------------------------------------------ discovery + + /// + /// The sys catalog views rather than INFORMATION_SCHEMA. The standard views are defined to show + /// only what the caller has a permission on and cannot report an estimated row count at all; + /// sys.objects joined to the partition stats answers both in one query. + /// + protected override async Task> DiscoverAsync(DbConnection connection, + DiscoverRequest request, CancellationToken cancellationToken) + { + var schema = request.Schema; + var like = request.NameLike?.ToLowerInvariant(); + + return request.ObjectType switch + { + "procedure" or "function" => + await RoutinesAsync(connection, request, schema, like, cancellationToken), + "sequence" => await SequencesAsync(connection, request, schema, like, cancellationToken), + _ => await RelationsAsync(connection, request, schema, like, cancellationToken) + }; + } + + async Task> RelationsAsync(DbConnection connection, DiscoverRequest request, + string schema, string like, CancellationToken cancellationToken) + { + // U user table, V view. Nothing else in sys.objects is a thing an integration reads rows + // from. + var wanted = request.ObjectType == "view" ? "V" : "U"; + + var parameters = new Dictionary { ["wanted"] = wanted }; + var sql = new StringBuilder(@" + select s.name as [schema], o.name, o.type, + cast(ep.value as nvarchar(max)) as comment, + (select sum(p.rows) from sys.partitions p + where p.object_id = o.object_id and p.index_id in (0, 1)) as estimate + from sys.objects o + join sys.schemas s on s.schema_id = o.schema_id + left join sys.extended_properties ep + on ep.major_id = o.object_id and ep.minor_id = 0 and ep.name = 'MS_Description' + where o.type = @wanted and o.is_ms_shipped = 0"); + + if (schema != null) { sql.Append(" and s.name = @schema"); parameters["schema"] = schema; } + if (like != null) { sql.Append(" and charindex(@nameLike, lower(o.name)) > 0"); parameters["nameLike"] = like; } + + // OFFSET/FETCH needs an ORDER BY, which this has anyway — the list is meant to be stable + // between pages. + sql.Append(" order by s.name, o.name offset @skip rows fetch next @take rows only"); + parameters["skip"] = request.Skip; + parameters["take"] = request.Take; + + var objects = new List(); + using (var command = connection.CreateCommand()) + { + command.CommandText = sql.ToString(); + command.CommandTimeout = Options.CommandTimeoutSeconds; + AddParameters(command, parameters); + + using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + objects.Add(new DbObject + { + Schema = reader.GetString(0), + Name = reader.GetString(1), + Type = reader.GetString(2).Trim() == "V" ? "view" : "table", + Comment = reader.IsDBNull(3) ? null : reader.GetString(3), + + // The heap or clustered index row count from the partition stats: maintained by + // the engine and cheap to read. Reported as an estimate everywhere it surfaces + // — the alternative is COUNT(*) on a stranger's table, which a menu should not + // do. Null for a view, which has no partitions of its own. + RowCount = !request.IncludeRowCounts || reader.IsDBNull(4) + ? null + : Math.Max(0, reader.GetInt64(4)) + }); + } + + if (request.IncludeColumns && objects.Count > 0) + await FillColumnsAsync(connection, objects, cancellationToken); + + return objects; + } + + async Task FillColumnsAsync(DbConnection connection, List objects, + CancellationToken cancellationToken) + { + // One query for the whole page, not one per object: a page of 200 tables would otherwise + // be 200 round trips, and against a remote database that is the difference between a + // screen that opens and one that times out. + var pairs = PairsOf(objects.Select(o => (o.Schema, o.Name))); + + using var command = connection.CreateCommand(); + command.CommandText = $@" + select s.name as [schema], o.name as [object], c.name as [column], + t.name as type_name, c.max_length, c.precision, c.scale, + c.is_nullable, c.column_id, + c.is_identity | c.is_computed | + (case when c.default_object_id <> 0 then 1 else 0 end) as generated, + (case when exists ( + select 1 + from sys.index_columns ic + join sys.indexes i + on i.object_id = ic.object_id and i.index_id = ic.index_id + where ic.object_id = c.object_id and ic.column_id = c.column_id + and i.is_primary_key = 1) then 1 else 0 end) as is_pk + from sys.columns c + join sys.objects o on o.object_id = c.object_id + join sys.schemas s on s.schema_id = o.schema_id + join sys.types t on t.user_type_id = c.user_type_id + and exists (select 1 from {pairs} as w(s, o) + where w.s = s.name collate database_default + and w.o = o.name collate database_default) + order by s.name, o.name, c.column_id"; + command.CommandTimeout = Options.CommandTimeoutSeconds; + + var byObject = objects.ToDictionary(o => $"{o.Schema}.{o.Name}"); + + using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + var key = $"{reader.GetString(0)}.{reader.GetString(1)}"; + if (!byObject.TryGetValue(key, out var target)) continue; + + var typeName = reader.GetString(3); + + // max_length is in BYTES, and an nvarchar stores two per character — so the number a + // schema was declared with is half of it. -1 is the max/blob form. + var maxLength = reader.GetInt16(4); + var wide = typeName.StartsWith("n", StringComparison.OrdinalIgnoreCase) + && typeName.IndexOf("char", StringComparison.OrdinalIgnoreCase) >= 0; + + target.Columns.Add(new DbColumn + { + Name = reader.GetString(2), + DbType = DescribeType(typeName, maxLength, reader.GetByte(5), reader.GetByte(6), wide), + ClrType = ClrTypeOf(typeName), + Nullable = reader.GetBoolean(7), + Ordinal = reader.GetInt32(8), + Generated = reader.GetInt32(9) != 0, + PrimaryKey = reader.GetInt32(10) != 0, + Length = maxLength < 0 ? null : wide ? maxLength / 2 : maxLength, + Precision = reader.GetByte(5), + Scale = reader.GetByte(6) + }); + } + } + + /// The type as a schema would declare it, which is what an operator recognises. + static string DescribeType(string typeName, short maxLength, byte precision, byte scale, bool wide) + { + var bare = typeName.ToLowerInvariant(); + + if (bare is "decimal" or "numeric") return $"{bare}({precision},{scale})"; + if (bare is "datetime2" or "time" or "datetimeoffset") return $"{bare}({scale})"; + + if (bare.EndsWith("char") || bare.EndsWith("binary")) + return maxLength < 0 ? $"{bare}(max)" : $"{bare}({(wide ? maxLength / 2 : maxLength)})"; + + return bare; + } + + async Task> RoutinesAsync(DbConnection connection, DiscoverRequest request, + string schema, string like, CancellationToken cancellationToken) + { + // P stored procedure; FN scalar, IF inline table-valued, TF multi-statement table-valued. + // All three function forms are worth listing — the table-valued ones are what a receive + // statement would select from. + var types = request.ObjectType == "procedure" + ? new[] { "P" } + : new[] { "FN", "IF", "TF" }; + + var parameters = new Dictionary(); + var placeholders = new List(); + for (var i = 0; i < types.Length; i++) + { + placeholders.Add($"@type{i}"); + parameters[$"type{i}"] = types[i]; + } + + var sql = new StringBuilder($@" + select s.name as [schema], o.name, o.type, + cast(ep.value as nvarchar(max)) as comment + from sys.objects o + join sys.schemas s on s.schema_id = o.schema_id + left join sys.extended_properties ep + on ep.major_id = o.object_id and ep.minor_id = 0 and ep.name = 'MS_Description' + where o.type in ({string.Join(",", placeholders)}) and o.is_ms_shipped = 0"); + + if (schema != null) { sql.Append(" and s.name = @schema"); parameters["schema"] = schema; } + if (like != null) { sql.Append(" and charindex(@nameLike, lower(o.name)) > 0"); parameters["nameLike"] = like; } + + sql.Append(" order by s.name, o.name offset @skip rows fetch next @take rows only"); + parameters["skip"] = request.Skip; + parameters["take"] = request.Take; + + var objects = new List(); + using (var command = connection.CreateCommand()) + { + command.CommandText = sql.ToString(); + command.CommandTimeout = Options.CommandTimeoutSeconds; + AddParameters(command, parameters); + + using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + objects.Add(new DbObject + { + Schema = reader.GetString(0), + Name = reader.GetString(1), + Type = reader.GetString(2).Trim() == "P" ? "procedure" : "function", + Comment = reader.IsDBNull(3) ? null : reader.GetString(3) + }); + } + + if (objects.Count > 0) await FillParametersAsync(connection, objects, cancellationToken); + return objects; + } + + async Task FillParametersAsync(DbConnection connection, List routines, + CancellationToken cancellationToken) + { + var pairs = PairsOf(routines.Select(r => (r.Schema, r.Name))); + + using var command = connection.CreateCommand(); + command.CommandText = $@" + select s.name as [schema], o.name as [object], p.name as parameter, + t.name as type_name, p.is_output, p.parameter_id, p.max_length, p.precision, p.scale + from sys.parameters p + join sys.objects o on o.object_id = p.object_id + join sys.schemas s on s.schema_id = o.schema_id + join sys.types t on t.user_type_id = p.user_type_id + where exists (select 1 from {pairs} as w(s, o) + where w.s = s.name collate database_default + and w.o = o.name collate database_default) + order by s.name, o.name, p.parameter_id"; + command.CommandTimeout = Options.CommandTimeoutSeconds; + + var byRoutine = routines.ToDictionary(r => $"{r.Schema}.{r.Name}"); + + using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + var key = $"{reader.GetString(0)}.{reader.GetString(1)}"; + if (!byRoutine.TryGetValue(key, out var target)) continue; + + // parameter_id 0 is a scalar function's return value, and it has no name. + var ordinal = reader.GetInt32(5); + var name = reader.IsDBNull(2) || reader.GetString(2).Length == 0 + ? (ordinal == 0 ? "(returns)" : $"@p{ordinal}") + : reader.GetString(2).TrimStart('@'); + + var typeName = reader.GetString(3); + var wide = typeName.StartsWith("n", StringComparison.OrdinalIgnoreCase) + && typeName.IndexOf("char", StringComparison.OrdinalIgnoreCase) >= 0; + + target.Parameters.Add(new DbRoutineParameter + { + Name = name, + DbType = DescribeType(typeName, reader.GetInt16(6), reader.GetByte(7), reader.GetByte(8), wide), + Direction = ordinal == 0 ? "ReturnValue" : reader.GetBoolean(4) ? "Out" : "In", + Ordinal = ordinal + }); + } + } + + async Task> SequencesAsync(DbConnection connection, DiscoverRequest request, + string schema, string like, CancellationToken cancellationToken) + { + var parameters = new Dictionary(); + var sql = new StringBuilder(@" + select s.name as [schema], q.name, cast(q.current_value as bigint) as current_value + from sys.sequences q + join sys.schemas s on s.schema_id = q.schema_id + where q.is_ms_shipped = 0"); + + if (schema != null) { sql.Append(" and s.name = @schema"); parameters["schema"] = schema; } + if (like != null) { sql.Append(" and charindex(@nameLike, lower(q.name)) > 0"); parameters["nameLike"] = like; } + + sql.Append(" order by s.name, q.name offset @skip rows fetch next @take rows only"); + parameters["skip"] = request.Skip; + parameters["take"] = request.Take; + + var objects = new List(); + using var command = connection.CreateCommand(); + command.CommandText = sql.ToString(); + command.CommandTimeout = Options.CommandTimeoutSeconds; + AddParameters(command, parameters); + + using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + objects.Add(new DbObject + { + Schema = reader.GetString(0), + Name = reader.GetString(1), + Type = "sequence", + + // Null until the sequence has been used at all, which is a meaningful answer of its + // own: nothing has drawn from this counter yet. + RowCount = reader.IsDBNull(2) ? null : reader.GetInt64(2) + }); + + return objects; + } + + /// + /// The page's (schema, object) pairs as a table to join against. + /// + /// SQL Server has no row constructor in IN — where (a, b) in ((..),(..)) is PostgreSQL + /// and MySQL syntax and is a parse error here — so the pairs become a VALUES table and the + /// filter becomes an EXISTS against it. The collation is forced to the database's own because + /// a VALUES literal takes the server default, and on an instance whose default differs from + /// the database's the comparison fails outright rather than merely matching oddly. + /// + static string PairsOf(IEnumerable<(string Schema, string Name)> objects) => + "(values " + string.Join(",", objects.Select(o => $"({Quote(o.Schema)},{Quote(o.Name)})")) + ")"; + + /// + /// An N-prefixed literal for the pairs above. Doubling the quote is SQL Server's own + /// escape, and these are identifiers this connection just read back from the catalog — quoted + /// anyway, because "it cannot contain a quote" is the kind of assumption that survives right up + /// until a table is named oddly. + /// + static string Quote(string value) => "N'" + (value ?? "").Replace("'", "''") + "'"; + + /// + /// What a value of this column arrives as in a result row. Coarse on purpose — it tells a + /// mapper author whether to expect a string or a number, and is not trying to be a type system. + /// + static string ClrTypeOf(string typeName) => typeName.ToLowerInvariant() switch + { + "bit" => "bool", + "tinyint" or "smallint" or "int" => "int", + "bigint" => "long", + "decimal" or "numeric" or "money" or "smallmoney" => "decimal", + "float" or "real" => "double", + "date" or "datetime" or "datetime2" or "smalldatetime" => "DateTime", + "datetimeoffset" => "DateTimeOffset", + "time" => "TimeSpan", + "uniqueidentifier" => "Guid", + "binary" or "varbinary" or "image" or "timestamp" or "rowversion" => "byte[]", + "xml" => "string", + _ => "string" + }; + + void AddParameters(DbCommand command, Dictionary parameters) + { + foreach (var kv in parameters) + { + var parameter = command.CreateParameter(); + parameter.ParameterName = kv.Key; + parameter.Value = kv.Value ?? DBNull.Value; + command.Parameters.Add(parameter); + } + } +} diff --git a/SW.Bitween.Adapters.Db.SqlServer/SqlServerOptions.cs b/SW.Bitween.Adapters.Db.SqlServer/SqlServerOptions.cs new file mode 100644 index 00000000..33b6fe09 --- /dev/null +++ b/SW.Bitween.Adapters.Db.SqlServer/SqlServerOptions.cs @@ -0,0 +1,80 @@ +using SW.Bitween.Adapters; + +namespace SW.Bitween.Adapters.Db.SqlServer; + +/// +/// Every setting here arrives as a DataSource property, bound by name, and the form an operator +/// fills in is generated from these attributes — so a field added here appears in Bitween with no +/// front-end change. +/// +/// The hints spend their weight on encryption, because SQL Server is the engine where the default +/// changed underneath everybody: the modern driver encrypts by default and then refuses a +/// self-signed certificate, which is what most on-premises instances present. +/// +[AdapterSettings( + Kind = "Relational", + Label = "SQL Server", + Description = "A Microsoft SQL Server or Azure SQL database, held open with a pooled " + + "connection so statements, procedures and polling receivers do not pay a " + + "connect on every message.")] +public class SqlServerOptions : DbOptionsBase +{ + [AdapterSetting(Required = true, Hint = + "Host name, IP, or host\\instance for a named instance. No tcp: prefix — set Port instead.")] + public string Host { get; set; } = "localhost"; + + [AdapterSetting(Default = "1433", Hint = + "Ignored when Host names an instance: a named instance is resolved through the SQL Server " + + "Browser rather than by port.")] + public int Port { get; set; } = 1433; + + [AdapterSetting(Required = true, Hint = "The database to connect to, not the server.")] + public string Database { get; set; } + + [AdapterSetting(Required = true, Hint = + "A SQL login, or DOMAIN\\user for Windows authentication where the host supports it.")] + public string UserName { get; set; } + + [AdapterSetting(Secret = true, Required = true)] + public string Password { get; set; } + + /// + /// The default schema for unqualified names is a property of the LOGIN, not of the connection + /// string — there is no equivalent of PostgreSQL's search_path to set here. This runs a + /// per-connection default instead, which is why it is worded as what it does rather than as a + /// setting name. + /// + [AdapterSetting(Hint = + "Run as this schema, so an unqualified table name in a statement resolves against it. " + + "Leave empty to use the login's own default, which is usually dbo. Only takes effect if " + + "the login can impersonate the schema's owner; when it cannot, the connection still " + + "works and unqualified names keep resolving as before.")] + public string Schema { get; set; } + + [AdapterSetting(Default = "true", AllowedValues = new[] { "true", "false" }, Hint = + "Encrypt the connection. On by default in the modern driver, which is a change from the " + + "old one — and the reason an instance that worked for years starts failing on upgrade.")] + public bool Encrypt { get; set; } = true; + + [AdapterSetting(Default = "false", AllowedValues = new[] { "true", "false" }, Hint = + "Accept the server's certificate without validating it. Needed for the usual on-premises " + + "instance with a self-signed certificate — and it means an attacker between here and the " + + "server could present their own. Prefer installing the certificate; use this knowingly.")] + public bool TrustServerCertificate { get; set; } + + [AdapterSetting(Default = "Bitween", Hint = + "What this connection calls itself in sys.dm_exec_sessions. Worth keeping distinctive: it " + + "is how a DBA works out which of the connections on their server is yours.")] + public string ApplicationName { get; set; } = "Bitween"; + + [AdapterSetting(Default = "false", AllowedValues = new[] { "true", "false" }, Hint = + "Use MultipleActiveResultSets. Off unless something needs it: it lets one connection hold " + + "several open readers, and the cost is that the connection cannot be reset cleanly " + + "between uses, which is the opposite of what a shared pool wants.")] + public bool MultipleActiveResultSets { get; set; } + + [AdapterSetting(Default = "false", AllowedValues = new[] { "true", "false" }, Hint = + "Read committed snapshot for this connection's transactions, so readers do not block " + + "writers. Only has an effect where the database has snapshot isolation enabled.")] + public bool SnapshotIsolation { get; set; } +} diff --git a/SW.Bitween.Api/Resources/DataSourceStatements/Create.cs b/SW.Bitween.Api/Resources/DataSourceStatements/Create.cs index b73ab514..06ac374c 100644 --- a/SW.Bitween.Api/Resources/DataSourceStatements/Create.cs +++ b/SW.Bitween.Api/Resources/DataSourceStatements/Create.cs @@ -42,7 +42,7 @@ await requestContext.EnsurePermission(dbContext, + "something to a Relational one."); await EnsureNameIsFree(dbContext, dataSourceId, model.Name, existingId: null); - await EnsureSqlIsValid(validator, dataSourceId, model.Sql); + var checkedSql = await EnsureSqlIsValid(validator, dataSourceId, model.Sql); var entity = new DataSourceStatement { @@ -58,7 +58,10 @@ await requestContext.EnsurePermission(dbContext, dbContext.Add(entity); await dbContext.SaveChangesAsync(); - return entity.Id; + // The id, and whether the SQL was actually checked. Saying so matters because validation + // is best-effort: when the adapter is not running there is nobody to ask, and a save that + // reported nothing would look exactly like one that had been verified. + return new { Id = entity.Id, Checked = checkedSql }; } /// @@ -69,12 +72,15 @@ await requestContext.EnsurePermission(dbContext, /// connection, and blocking someone from saving a fix because the thing they are fixing it /// for is broken would be exactly backwards. /// - internal static async Task EnsureSqlIsValid( + /// True when the database actually looked at it; false when nobody could be asked. + internal static async Task EnsureSqlIsValid( SW.Bitween.Services.DataSources.StatementValidator validator, int dataSourceId, string sql) { var result = await validator.ValidateAsync(dataSourceId, sql); if (result.Checked && !result.Ok) throw new SWException($"The database will not accept this statement. {result.Error}"); + + return result.Checked; } /// diff --git a/SW.Bitween.Api/Resources/DataSourceStatements/Update.cs b/SW.Bitween.Api/Resources/DataSourceStatements/Update.cs index 9ca659e1..403da43b 100644 --- a/SW.Bitween.Api/Resources/DataSourceStatements/Update.cs +++ b/SW.Bitween.Api/Resources/DataSourceStatements/Update.cs @@ -43,8 +43,11 @@ public async Task Handle(int key, DataSourceStatementUpdate model) // Only when it actually changed: re-checking untouched SQL would refuse a rename, or a // change of owner, because of a table someone dropped last week — a fault worth surfacing // but not here, and not as a block on an unrelated edit. + // True when the SQL did not change: there was nothing to check, which is not the same as + // "could not check" and should not warn as though it were. + var checkedSql = true; if (!string.Equals(entity.Sql, model.Sql, System.StringComparison.Ordinal)) - await Create.EnsureSqlIsValid(validator, entity.DataSourceId, model.Sql); + checkedSql = await Create.EnsureSqlIsValid(validator, entity.DataSourceId, model.Sql); entity.Name = model.Name.Trim(); entity.Sql = model.Sql; @@ -55,7 +58,7 @@ public async Task Handle(int key, DataSourceStatementUpdate model) entity.Inactive = model.Inactive; await dbContext.SaveChangesAsync(); - return entity.Id; + return new { Id = entity.Id, Checked = checkedSql }; } private class Validate : AbstractValidator diff --git a/SW.Bitween.Api/Resources/Subscriptions/Get.cs b/SW.Bitween.Api/Resources/Subscriptions/Get.cs index 24847542..5672ae76 100644 --- a/SW.Bitween.Api/Resources/Subscriptions/Get.cs +++ b/SW.Bitween.Api/Resources/Subscriptions/Get.cs @@ -41,6 +41,12 @@ public async Task Handle(int key) Inactive = subscriber.Inactive, MapperId = subscriber.MapperId, ReceiverId = subscriber.ReceiverId, + + // Which data source every stage of this subscription runs through. Dropped + // from this projection once, and the cost was quiet: the UI read every bound + // subscription as unbound, so opening one and saving it cleared the binding. + DataSourceId = subscriber.DataSourceId, + Name = subscriber.Name, PartnerId = subscriber.PartnerId, MapperProperties = subscriber.MapperProperties.ToKeyAndValueCollection(), diff --git a/SW.Bitween.Api/Services/Adapters/ResidentAdapterRuntime.cs b/SW.Bitween.Api/Services/Adapters/ResidentAdapterRuntime.cs index cd67b5dc..15b0e6f2 100644 --- a/SW.Bitween.Api/Services/Adapters/ResidentAdapterRuntime.cs +++ b/SW.Bitween.Api/Services/Adapters/ResidentAdapterRuntime.cs @@ -1,8 +1,10 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using SW.Bitween.Domain.DataSources; using SW.PrimitiveTypes; using SW.Serverless; using SW.Serverless.Resident; @@ -75,19 +77,40 @@ public async Task BeginAsync(string adapterId, AdapterRole role if (dataSourceId != null) { var running = adapters.Get(adapterId, dataSourceId); - if (running == null) - throw new BitweenException( - $"Data source {dataSourceId} is not running on this node, so adapter " - + $"'{adapterId}' has no connection to work through. If the data source is " - + "exclusive, another node holds it; if it is per-node, look at its health — " - + "the supervisor could not start it here."); - - // The subscription's own adapter properties travel with each CALL, not with the - // process: this instance is shared by every subscription bound to the data source, and - // its startup values are the data source's. Without this a subscription could not say - // which statement to run — it would be reading whatever the data source was started - // with, which is the same answer for all of them. - return new RunningInstanceSession(running, spec.StartupValues); + + if (running != null) + // The subscription's own adapter properties travel with each CALL, not with the + // process: this instance is shared by every subscription bound to the data source, + // and its startup values are the data source's. Without this a subscription could + // not say which statement to run — it would be reading whatever the data source + // was started with, which is the same answer for all of them. + return new RunningInstanceSession(running, spec.StartupValues); + + // Not here. For an EXCLUSIVE data source that is the normal case on every node but + // one — a broker connection is held by a single node so that a queue is drained once. + // + // Exclusivity is about CONSUMING, though, not about connecting. A subscription's + // handler runs on whichever node picked up the message, so a delivery that publishes + // to the customer's broker would fail on every node but the owner — which is to say, + // almost always. A publish-only connection of our own is the answer: it sends and + // never subscribes, so nothing is consumed twice. + if (role is AdapterRole.Handler or AdapterRole.Mapper) + { + var publishing = await PublishOnlySpecAsync(adapterId, dataSourceId); + if (publishing != null) + // The same split as the exclusive path: the rented instance's startup values + // are the CONNECTION's, and the slot's own properties — where to publish above + // all — travel with the call. Passing only the spec would send the message to + // an endpoint the connection never knew about, which is to say nowhere. + return new ResidentAdapterSession( + await adapters.RentAsync(publishing), spec.StartupValues); + } + + throw new BitweenException( + $"Data source {dataSourceId} is not running on this node, so adapter " + + $"'{adapterId}' has no connection to work through. If the data source is " + + "exclusive, another node holds it; if it is per-node, look at its health — " + + "the supervisor could not start it here."); } // Rented, not started: the process is already up, so the call costs a round trip rather @@ -96,6 +119,56 @@ public async Task BeginAsync(string adapterId, AdapterRole role return new ResidentAdapterSession(await adapters.RentAsync(spec)); } + /// + /// A spec for a connection that can SEND but will never consume, built from the data source's + /// own settings. + /// + /// Consume=false is the whole point, and it is the same switch the connection test uses + /// for the same reason: a second instance that subscribed would drain the customer's queue + /// alongside the node that owns it, and the lease exists precisely to stop that. + /// + /// Pooled rather than exclusive, so several nodes may hold one at once — which is correct for + /// publishing and wrong for consuming. The pool key includes a hash of these values, so the + /// publish-only instance is a different renter from anything else and cannot be handed the + /// consuming one by mistake. + /// + /// Null when there is no such data source, or it is not a broker: a relational source is + /// per-node and is expected to be here, so "not running" is a fault to report rather than + /// something to work around. + /// + private async Task PublishOnlySpecAsync(string adapterId, string dataSourceId) + { + if (!int.TryParse(dataSourceId, out var id)) return null; + + var dbContext = serviceProvider.GetService(); + if (dbContext == null) return null; + + var dataSource = await dbContext.Set().AsNoTracking() + .FirstOrDefaultAsync(d => d.Id == id); + + if (dataSource == null || dataSource.Kind != DataSourceKind.Broker) return null; + + var spec = new AdapterSpec { AdapterId = adapterId }; + + foreach (var kv in dataSource.Properties ?? new Dictionary()) + spec.StartupValues[kv.Key] = kv.Value; + + // Never subscribe, and carry no endpoints to subscribe to even if something ignored the + // flag. Where to PUBLISH travels with the call, not with the process. + spec.StartupValues["Consume"] = "false"; + spec.StartupValues.Remove("Endpoints"); + + // Distinct from any other renter of this adapter, so a publish-only instance is never + // confused with one somebody else configured. + spec.PoolKey = $"{adapterId}:publish:{id}"; + + logger.LogDebug( + "Data source {DataSourceId} is owned elsewhere; publishing through a send-only " + + "connection on this node.", id); + + return spec; + } + /// /// A session against the data source's long-lived instance. Nothing is returned on dispose: /// the instance is not ours to give back, and it must outlive this Xchange to be any use to @@ -114,13 +187,14 @@ public Task InvokeAsync(string method, object argument = null) => public ValueTask DisposeAsync() => default; } - private sealed class ResidentAdapterSession(IAdapterLease lease) : IAdapterSession + private sealed class ResidentAdapterSession( + IAdapterLease lease, IDictionary properties = null) : IAdapterSession { public Task InvokeAsync(string method, object argument = null) => - lease.InvokeAsync(method, argument); + lease.InvokeAsync(method, argument, properties: properties); public Task InvokeAsync(string method, object argument = null) => - lease.InvokeAsync(method, argument); + lease.InvokeAsync(method, argument, properties: properties); public ValueTask DisposeAsync() => lease.DisposeAsync(); } diff --git a/SW.Bitween.Api/Services/DataSources/BusProviderSupervisor.cs b/SW.Bitween.Api/Services/DataSources/BusProviderSupervisor.cs index fc48a097..9f330338 100644 --- a/SW.Bitween.Api/Services/DataSources/BusProviderSupervisor.cs +++ b/SW.Bitween.Api/Services/DataSources/BusProviderSupervisor.cs @@ -343,10 +343,22 @@ private async Task RecordStartFailureAsync(BitweenDbContext dbContext, DataSourc /// private async Task WriteBackHealthAsync(BitweenDbContext dbContext, CancellationToken cancellationToken) { - var health = adapters.Describe().ToDictionary(h => h.InstanceKey); + // Only the instances that ARE a data source. Describe() answers for everything this host + // holds, and a POOLED instance is keyed by its pool slot rather than by a data source — + // several can share one key, which made a plain ToDictionary throw "an item with the same + // key has already been added" and take the whole reconcile pass down with it. + // + // Pooled bus instances became routine when a delivery on a node that does not own the + // connection started opening a send-only one. They have no health to write back: nothing + // owns them, and the row they would write to belongs to the node that does. + var health = adapters.Describe() + .Where(h => int.TryParse(h.InstanceKey, out var id) && id > 0) + .GroupBy(h => h.InstanceKey) + .ToDictionary(g => g.Key, g => g.First()); + if (health.Count == 0) return; - var ids = health.Keys.Select(k => int.TryParse(k, out var id) ? id : 0).Where(i => i > 0).ToList(); + var ids = health.Keys.Select(int.Parse).ToList(); var rows = await dbContext.Set() .Where(d => ids.Contains(d.Id)) .ToListAsync(cancellationToken); diff --git a/SW.Bitween.IntegrationTests/Fixtures/BitweenFixture.cs b/SW.Bitween.IntegrationTests/Fixtures/BitweenFixture.cs index 0f8c3fda..ffd74b51 100644 --- a/SW.Bitween.IntegrationTests/Fixtures/BitweenFixture.cs +++ b/SW.Bitween.IntegrationTests/Fixtures/BitweenFixture.cs @@ -322,6 +322,14 @@ await AdapterInstaller.InstallAsync(cloudFiles, "SW.Bitween.Adapters.Db.PostgreSql", BusAdapters.PostgreSql, "SW.Bitween.Adapters.Db.PostgreSql.dll", new Dictionary { ["Protocol"] = "2", ["Lifecycle"] = "resident" }); + await AdapterInstaller.InstallAsync(cloudFiles, + "SW.Bitween.Adapters.Db.MySql", BusAdapters.MySql, + "SW.Bitween.Adapters.Db.MySql.dll", + new Dictionary { ["Protocol"] = "2", ["Lifecycle"] = "resident" }); + await AdapterInstaller.InstallAsync(cloudFiles, + "SW.Bitween.Adapters.Db.SqlServer", BusAdapters.SqlServer, + "SW.Bitween.Adapters.Db.SqlServer.dll", + new Dictionary { ["Protocol"] = "2", ["Lifecycle"] = "resident" }); } await App.StartAsync(); diff --git a/SW.Bitween.IntegrationTests/Fixtures/BusAdapters.cs b/SW.Bitween.IntegrationTests/Fixtures/BusAdapters.cs index 231d0653..84957069 100644 --- a/SW.Bitween.IntegrationTests/Fixtures/BusAdapters.cs +++ b/SW.Bitween.IntegrationTests/Fixtures/BusAdapters.cs @@ -12,4 +12,6 @@ public static class BusAdapters /// public const string Oracle = "bitween.db.oracle"; public const string PostgreSql = "bitween.db.postgresql"; + public const string MySql = "bitween.db.mysql"; + public const string SqlServer = "bitween.db.sqlserver"; } diff --git a/SW.Bitween.IntegrationTests/Fixtures/MySqlDbFixture.cs b/SW.Bitween.IntegrationTests/Fixtures/MySqlDbFixture.cs new file mode 100644 index 00000000..faab1d56 --- /dev/null +++ b/SW.Bitween.IntegrationTests/Fixtures/MySqlDbFixture.cs @@ -0,0 +1,125 @@ +using System; +using System.Threading.Tasks; +using MySqlConnector; +using Testcontainers.MySql; +using Xunit; + +namespace SW.Bitween.IntegrationTests.Fixtures; + +/// +/// A MySQL for the adapter tests to point at — deliberately its own container rather than the one +/// runs for the application database. +/// +/// A class fixture, not on the test class: xUnit builds a new instance +/// of a test class per test method, so a container started there is a container per test. +/// +/// The seed mirrors the PostgreSQL one object for object, because the point of having both is that +/// the shared core behaves identically and the differences show up where they are real — here, a +/// procedure that returns rows by SELECTing, which is the thing PostgreSQL cannot do. +/// +public class MySqlDbFixture : IAsyncLifetime +{ + public const string Table = "bitween_orders"; + + MySqlContainer _container; + + /// Set when Docker is unavailable, so the tests skip rather than fail. + public string Unavailable { get; private set; } + + public string Host => _container.Hostname; + public int Port => _container.GetMappedPublicPort(3306); + public string Database => "bitween_adapter"; + public string User => "bitween"; + public string Password => "bitween_pw"; + + public async Task InitializeAsync() + { + try + { + _container = new MySqlBuilder() + .WithImage("mysql:8.4") + .WithDatabase(Database) + .WithUsername(User) + .WithPassword(Password) + + // Creating a FUNCTION requires SUPER while binary logging is on, because a + // non-deterministic one would make the binary log unsafe to replay. Our user is + // not SUPER and should not be, so the server is told to trust function creators — + // which is the switch a DBA sets for exactly this, and is why the adapter's own + // capability list does not promise that creating routines is something it can do. + .WithCommand("--log-bin-trust-function-creators=1") + .Build(); + + await _container.StartAsync(); + await SeedAsync(); + } + catch (Exception ex) + { + Unavailable = ex.Message; + } + } + + public async Task DisposeAsync() + { + if (_container != null) await _container.DisposeAsync(); + } + + public string AdminConnectionString => + $"Server={Host};Port={Port};Database={Database};User ID={User};Password={Password};" + + "AllowUserVariables=true;AllowPublicKeyRetrieval=true"; + + async Task SeedAsync() + { + await using var connection = new MySqlConnection(AdminConnectionString); + await connection.OpenAsync(); + + await ExecuteAsync(connection, $@" + create table {Table} ( + id int primary key, + customer varchar(50), + amount decimal(10,2), + created_at timestamp not null default current_timestamp, + processed tinyint(1) not null default 0 + ) comment 'Customer orders'"); + + for (var i = 1; i <= 25; i++) + await ExecuteAsync(connection, + $"insert into {Table} (id, customer, amount) " + + $"values ({i}, '{(i % 2 == 0 ? "acme" : "globex")}', {i * 10}.50)"); + + // A procedure that returns rows simply by SELECTing — no cursor to declare, nothing to + // bind. This is the capability PostgreSQL declares false and MySQL declares true, and the + // reason the two adapters are worth testing against the same shape of schema. + await ExecuteAsync(connection, $@" + create procedure orders_by_customer(in p_customer varchar(50)) + begin + select * from {Table} where customer = p_customer order by id; + end"); + + // A procedure that returns a value through an OUT parameter rather than a result set. + await ExecuteAsync(connection, $@" + create procedure count_orders(in p_customer varchar(50), out p_total int) + begin + select count(*) into p_total from {Table} where customer = p_customer; + end"); + + // And a scalar function, which is a different object type in the catalog. + await ExecuteAsync(connection, $@" + create function total_for(p_customer varchar(50)) + returns decimal(12,2) + deterministic + reads sql data + begin + declare v_total decimal(12,2); + select coalesce(sum(amount), 0) into v_total from {Table} where customer = p_customer; + return v_total; + end"); + } + + static async Task ExecuteAsync(MySqlConnection connection, string sql) + { + await using var command = connection.CreateCommand(); + command.CommandText = sql; + await command.ExecuteNonQueryAsync(); + } +} diff --git a/SW.Bitween.IntegrationTests/Fixtures/SqlServerDbFixture.cs b/SW.Bitween.IntegrationTests/Fixtures/SqlServerDbFixture.cs new file mode 100644 index 00000000..75cb4928 --- /dev/null +++ b/SW.Bitween.IntegrationTests/Fixtures/SqlServerDbFixture.cs @@ -0,0 +1,184 @@ +using System; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient; +using DotNet.Testcontainers.Builders; +using Testcontainers.MsSql; +using Xunit; + +namespace SW.Bitween.IntegrationTests.Fixtures; + +/// +/// A SQL Server for the adapter tests to point at — deliberately its own container rather than the +/// one runs for the application database. +/// +/// A class fixture, not on the test class: xUnit builds a new instance +/// of a test class per test method, so a container started there is a container per test. +/// +/// The image is the 2022 developer edition, which is what Testcontainers defaults to and what runs +/// on both x64 and, through emulation, Apple silicon. It is around 1.5 GB — an order of magnitude +/// less than Oracle's, so unlike those these run on every pass rather than nightly. +/// +public class SqlServerDbFixture : IAsyncLifetime +{ + public const string Table = "bitween_orders"; + + MsSqlContainer _container; + + /// Set when Docker is unavailable, so the tests skip rather than fail. + public string Unavailable { get; private set; } + + public string Host => _container.Hostname; + public int Port => _container.GetMappedPublicPort(1433); + + /// + /// A database of our own rather than the container's master. Discovery filters by schema and + /// master is full of the engine's own objects, so a test asserting on what it can see would be + /// asserting about Microsoft's schema as much as ours. + /// + public string Database => "bitween_adapter"; + + public string User => "sa"; + public string Password => "yourStrong(!)Password"; + + public async Task InitializeAsync() + { + try + { + _container = new MsSqlBuilder() + // Pinned, and not to Testcontainers' default. That default is 2019, which has no + // arm64 image and no emulation path — it exits the moment it starts on an Apple + // silicon machine, which surfaces as "container is not running" rather than as + // anything about the architecture. 2022 is the first release Microsoft publishes + // for arm64, and it runs on both. + .WithImage("mcr.microsoft.com/mssql/server:2022-CU14-ubuntu-22.04") + .WithPassword(Password) + + // Wait on the PORT, not on the built-in sqlcmd probe. SQL Server listens well + // before it will accept a login, and under emulation on Apple silicon the gap is + // long enough that the default readiness check gives up and the container is torn + // down — which surfaces as "container is not running" rather than as a timeout. + .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(1433)) + .Build(); + + await _container.StartAsync(); + await WaitForLoginsAsync(); + await SeedAsync(); + } + catch (Exception ex) + { + Unavailable = ex.Message; + } + } + + public async Task DisposeAsync() + { + if (_container != null) await _container.DisposeAsync(); + } + + /// + /// TrustServerCertificate because the container presents a self-signed certificate, which is + /// the same reason the adapter offers the setting at all. + /// + public string AdminConnectionString => + $"Server={Host},{Port};Database={Database};User ID={User};Password={Password};" + + "Encrypt=True;TrustServerCertificate=True"; + + string MasterConnectionString => + $"Server={Host},{Port};Database=master;User ID={User};Password={Password};" + + "Encrypt=True;TrustServerCertificate=True"; + + /// + /// Polls until a login succeeds. The port opening only means the process is up; recovery of + /// master, msdb and tempdb runs after that, and a connection during it is refused with "not + /// currently available". Two minutes, because emulation is slow and a flaky skip reads as a + /// broken adapter. + /// + async Task WaitForLoginsAsync() + { + var deadline = DateTime.UtcNow.AddMinutes(2); + while (true) + { + try + { + await using var probe = new SqlConnection(MasterConnectionString); + await probe.OpenAsync(); + return; + } + catch when (DateTime.UtcNow < deadline) + { + await Task.Delay(TimeSpan.FromSeconds(3)); + } + } + } + + async Task SeedAsync() + { + await using (var master = new SqlConnection(MasterConnectionString)) + { + await master.OpenAsync(); + await ExecuteAsync(master, $"create database [{Database}]"); + } + + await using var connection = new SqlConnection(AdminConnectionString); + await connection.OpenAsync(); + + // A schema of its own, so the discovery tests can prove the schema filter does something — + // everything would otherwise be dbo, where a filter that did nothing would still pass. + await ExecuteAsync(connection, "create schema sales"); + + await ExecuteAsync(connection, $@" + create table sales.{Table} ( + id int not null primary key, + customer nvarchar(50) null, + amount decimal(10,2) null, + created_at datetime2(3) not null constraint df_created default sysutcdatetime(), + processed bit not null constraint df_processed default 0 + )"); + + await ExecuteAsync(connection, $@" + exec sys.sp_addextendedproperty + @name = N'MS_Description', @value = N'Customer orders', + @level0type = N'SCHEMA', @level0name = N'sales', + @level1type = N'TABLE', @level1name = N'{Table}'"); + + await ExecuteAsync(connection, "create sequence sales.bitween_order_seq as bigint start with 1000"); + + for (var i = 1; i <= 25; i++) + await ExecuteAsync(connection, + $"insert into sales.{Table} (id, customer, amount) " + + $"values ({i}, '{(i % 2 == 0 ? "acme" : "globex")}', {i * 10}.50)"); + + // A procedure that returns rows simply by SELECTing, like MySQL and unlike PostgreSQL. + await ExecuteAsync(connection, $@" + create procedure sales.orders_by_customer @p_customer nvarchar(50) + as + begin + set nocount on; + select * from sales.{Table} where customer = @p_customer order by id; + end"); + + // And one that answers through an OUT parameter instead. + await ExecuteAsync(connection, $@" + create procedure sales.count_orders @p_customer nvarchar(50), @p_total int output + as + begin + set nocount on; + select @p_total = count(*) from sales.{Table} where customer = @p_customer; + end"); + + // An inline table-valued function — the SQL Server shape a receive statement would select + // from, and a different object type in the catalog from a scalar one. + await ExecuteAsync(connection, $@" + create function sales.orders_for(@p_customer nvarchar(50)) + returns table + as + return (select * from sales.{Table} where customer = @p_customer)"); + } + + static async Task ExecuteAsync(SqlConnection connection, string sql) + { + await using var command = connection.CreateCommand(); + command.CommandText = sql; + await command.ExecuteNonQueryAsync(); + } +} diff --git a/SW.Bitween.IntegrationTests/SW.Bitween.IntegrationTests.csproj b/SW.Bitween.IntegrationTests/SW.Bitween.IntegrationTests.csproj index 41d03d76..385c44d2 100644 --- a/SW.Bitween.IntegrationTests/SW.Bitween.IntegrationTests.csproj +++ b/SW.Bitween.IntegrationTests/SW.Bitween.IntegrationTests.csproj @@ -26,8 +26,14 @@ + + + + + @@ -58,6 +64,8 @@ + + diff --git a/SW.Bitween.IntegrationTests/Tests/DataSourceStatementTests.cs b/SW.Bitween.IntegrationTests/Tests/DataSourceStatementTests.cs index c9e1aaea..cad79b0d 100644 --- a/SW.Bitween.IntegrationTests/Tests/DataSourceStatementTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/DataSourceStatementTests.cs @@ -259,8 +259,13 @@ async Task CreateStatementAsync(int dataSourceId, string name, string sql) var handler = ActivatorUtilities.CreateInstance( scope.ServiceProvider); - return (int)await handler.Handle( + // The handler answers with { Id, Checked } — the id, and whether the database actually + // looked at the SQL. Read through the property rather than cast, because the shape is an + // anonymous type and a cast to int is what this used to do. + var created = await handler.Handle( new DataSourceStatementCreate { DataSourceId = dataSourceId, Name = name, Sql = sql }); + + return (int)created.GetType().GetProperty("Id")!.GetValue(created)!; } async Task UpdateAsync(int id, DataSourceStatementUpdate model) diff --git a/SW.Bitween.IntegrationTests/Tests/ExternalBusGatewayTests.cs b/SW.Bitween.IntegrationTests/Tests/ExternalBusGatewayTests.cs index 8ff17e2d..b8120510 100644 --- a/SW.Bitween.IntegrationTests/Tests/ExternalBusGatewayTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/ExternalBusGatewayTests.cs @@ -6,6 +6,7 @@ using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using Newtonsoft.Json.Linq; using RabbitMQ.Client; using SW.Bitween.Domain; using SW.Bitween.Domain.DataSources; @@ -14,6 +15,7 @@ using SW.Bitween.Model; using SW.PrimitiveTypes; using SW.Serverless.Resident; +using SW.Bitween.Services.Adapters; using SW.Bitween.Services.DataSources; using Xunit; @@ -279,6 +281,106 @@ await WaitAsync(() => Depth(target) >= 1, TimeSpan.FromSeconds(15), "the published message never arrived on the external queue"); } + /// + /// A DELIVERY publishes, which is the half that was missing. Publish had existed since this + /// adapter did and nothing in the pipeline ever called it — the handler contract is Handle, + /// and this class had none — so an integration could drain a customer's queue and had no way + /// to answer on one. + /// + /// Driven the way the pipeline drives it: Handle, with the endpoint arriving as a + /// per-invocation property, because one instance serves every gateway on the broker and where + /// to send is the subscription's business rather than the connection's. + /// + [Fact] + public async Task A_delivery_publishes_the_message_it_was_given() + { + var consumed = Unique("handler-src"); + var target = Unique("handler-out"); + + // A queue the adapter is NOT consuming: depth on one it drains reads 0 whether the publish + // worked or not, because the message is taken as fast as it is sent. + var dataSourceId = await CreateDataSourceAsync(consumed, withGateway: false); + DeclareQueue(target); + + await using var adapter = await StartAsync(dataSourceId); + + var response = await adapter.Instance.InvokeAsync("Handle", + new XchangeFile("{\"delivered\":true}", "out.json"), + properties: new Dictionary { ["Endpoint"] = target }); + + await WaitAsync(() => Depth(target) >= 1, TimeSpan.FromSeconds(15), + "the delivered message never arrived on the external queue"); + + // The broker's receipt comes back as the response, so what was sent and under which id is + // on the exchange rather than only in a log. + Assert.Contains("messageId", response.Value("data") ?? response.ToString()); + } + + /// + /// A delivery with nowhere to send is refused where it was configured, rather than publishing + /// to a queue named the empty string. + /// + [Fact] + public async Task A_delivery_with_no_endpoint_says_so() + { + var dataSourceId = await CreateDataSourceAsync(Unique("handler-none"), withGateway: false); + await using var adapter = await StartAsync(dataSourceId); + + var error = await Assert.ThrowsAnyAsync(() => + adapter.Instance.InvokeAsync("Handle", + new XchangeFile("{}", "out.json"), + properties: new Dictionary())); + + Assert.Contains("no Endpoint and no Exchange", error.Message); + } + + /// + /// The multi-node case, and the reason this needed more than a Handle method. + /// + /// A broker data source is EXCLUSIVE: one node holds the connection so the customer's queue is + /// drained once. A subscription's delivery, though, runs on whichever node picked the message + /// up — so a publish would fail on every node but the owner, which is to say almost always. + /// + /// Exclusivity is about CONSUMING, not about connecting. When the owned instance is not here, + /// the runtime opens a send-only connection of its own — Consume=false, no endpoints, its own + /// pool key — and publishes through that. Nothing is consumed twice, and a delivery works + /// wherever it lands. + /// + /// Here nothing is running under the data source's instance key at all, which is exactly what + /// a non-owning node sees. + /// + [Fact] + public async Task A_delivery_publishes_from_a_node_that_does_not_own_the_connection() + { + var target = Unique("elsewhere-out"); + var dataSourceId = await CreateDataSourceAsync(Unique("elsewhere-src"), withGateway: false); + DeclareQueue(target); + + var host = fixture.App.Services.GetRequiredService(); + Assert.Null(host.Get(BusAdapters.RabbitMq, dataSourceId.ToString())); + + await using var scope = fixture.App.Services.CreateAsyncScope(); + var invoker = scope.ServiceProvider.GetRequiredService(); + + // Exactly what the pipeline passes: the data source id, and the slot's own properties. + var properties = new Dictionary + { + [StartupValuesFiller.DataSourceIdKey] = dataSourceId.ToString(), + ["Endpoint"] = target + }; + + await invoker.InvokeAsync(BusAdapters.RabbitMq, AdapterRole.Handler, + "Handle", new XchangeFile("{\"fromElsewhere\":true}", "out.json"), + properties, Guid.NewGuid().ToString("N")); + + await WaitAsync(() => Depth(target) >= 1, TimeSpan.FromSeconds(20), + "a node that does not own the connection could not publish"); + + // And it did NOT take ownership on the way past: the send-only instance is pooled under + // its own key, so the exclusive slot is still free for the node that should hold it. + Assert.Null(host.Get(BusAdapters.RabbitMq, dataSourceId.ToString())); + } + // ---------------------------------------------------------------- controls [Fact] diff --git a/SW.Bitween.IntegrationTests/Tests/MySqlAdapterTests.cs b/SW.Bitween.IntegrationTests/Tests/MySqlAdapterTests.cs new file mode 100644 index 00000000..fd86bd88 --- /dev/null +++ b/SW.Bitween.IntegrationTests/Tests/MySqlAdapterTests.cs @@ -0,0 +1,540 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Newtonsoft.Json.Linq; +using SW.Bitween.Domain.DataSources; +using SW.Bitween.IntegrationTests.Fixtures; +using SW.Serverless.Resident; +using Xunit; + +namespace SW.Bitween.IntegrationTests.Tests; + +/// +/// The MySQL data source provider, against a real MySQL. +/// +/// The same suite as PostgreSQL's and Oracle's, which is the point: three engines run on one core, +/// so the parts that are shared should behave identically and the parts that differ should differ +/// VISIBLY — in the capability list, and in how a routine returns rows. Everything here goes over +/// the resident transport into a real adapter process and out to a real database. +/// +[Collection("Bitween")] +public class MySqlAdapterTests : IClassFixture +{ + readonly BitweenFixture _fixture; + readonly MySqlDbFixture _mysql; + + public MySqlAdapterTests(BitweenFixture fixture, MySqlDbFixture mysql) + { + _fixture = fixture; + _mysql = mysql; + } + + static readonly SemaphoreSlim Gate = new(1, 1); + static int _dataSourceId; + + IResidentAdapterHost Host => _fixture.App.Services.GetRequiredService(); + + async Task AdapterAsync() + { + Skip.If(_mysql.Unavailable != null, $"MySQL is not available: {_mysql.Unavailable}"); + + await Gate.WaitAsync(); + try + { + if (_dataSourceId == 0) + { + _dataSourceId = await CreateDataSourceAsync(); + await StartAdapterAsync(); + } + } + finally + { + Gate.Release(); + } + + return Host.Get(BusAdapters.MySql, _dataSourceId.ToString()) + ?? throw new InvalidOperationException("The MySQL adapter is not running."); + } + + // ---------------------------------------------------------------- configuration + + [SkippableFact] + public async Task Connection_test_reports_each_stage_and_prepares_every_statement() + { + var adapter = await AdapterAsync(); + var result = await adapter.InvokeAsync("TestConnection", timeoutSeconds: 60); + + Assert.True(result.Value("ok"), result.ToString()); + + var steps = result["steps"]!.Select(s => s.Value("step")).ToList(); + Assert.Contains("connect", steps); + Assert.Contains("authenticate", steps); + Assert.Contains("privileges", steps); + Assert.Contains("statement:ordersForCustomer", steps); + Assert.All(result["steps"]!, s => Assert.True(s.Value("ok"), s.ToString())); + } + + /// + /// Where MySQL is honestly different from the other two. A procedure returns rows by SELECTing, + /// which PostgreSQL cannot do — but there is no MERGE, no RETURNING, no array type and no + /// sequence, and claiming any of them would have a statement written against it fail on a real + /// server. + /// + [SkippableFact] + public async Task Describe_reports_the_engine_and_where_it_differs() + { + var adapter = await AdapterAsync(); + var described = await adapter.InvokeAsync("Describe", timeoutSeconds: 60); + + Assert.Equal("MySQL", described.Value("engine")); + Assert.False(string.IsNullOrWhiteSpace(described.Value("serverVersion"))); + + // The capability PostgreSQL declares false and this one declares true. + Assert.True(described.Value("storedProcedures")); + Assert.True(described.Value("procedureResultSets")); + + // And the three it does not have. + Assert.False(described.Value("merge")); + Assert.False(described.Value("returning")); + Assert.False(described.Value("arrayTypes")); + + var objects = described["supportedObjects"]!.Select(o => o.Value()).ToList(); + Assert.Contains("procedure", objects); + Assert.Contains("function", objects); + // No sequences in MySQL — AUTO_INCREMENT belongs to a column, not to an object. + Assert.DoesNotContain("sequence", objects); + + // READ UNCOMMITTED does something here, unlike PostgreSQL where it is silently promoted. + var isolation = described["isolationLevels"]!.Select(i => i.Value()).ToList(); + Assert.Contains("ReadUncommitted", isolation); + + Assert.False(described.Value("logBasedCdc")); + Assert.False(described.Value("changeNotification")); + + Assert.NotEmpty(described["privileges"]!); + } + + // ---------------------------------------------------------------- discovery + + [SkippableFact] + public async Task Discover_finds_the_table_with_its_columns_and_key() + { + var adapter = await AdapterAsync(); + var result = await adapter.InvokeAsync("Discover", new + { + objectType = "table", + nameLike = MySqlDbFixture.Table, + includeColumns = true, + includeRowCounts = true + }, timeoutSeconds: 60); + + var table = result["objects"]!.Single(o => o.Value("name") == MySqlDbFixture.Table); + + // A schema IS a database in MySQL, and this is what that means in practice. + Assert.Equal(_mysql.Database, table.Value("schema")); + Assert.Equal("Customer orders", table.Value("comment")); + + var columns = table["columns"]!.ToList(); + Assert.Equal("id", columns[0].Value("name")); + Assert.True(columns[0].Value("primaryKey")); + Assert.False(columns[0].Value("nullable")); + + // tinyint(1) IS the boolean type in MySQL, and the driver returns one — so reporting "int" + // would mislead whoever writes the mapper. + var processed = columns.Single(c => c.Value("name") == "processed"); + Assert.Equal("bool", processed.Value("clrType")); + + // A CURRENT_TIMESTAMP default reports as DEFAULT_GENERATED in `extra`, which is a value the + // database fills in — the only distinction an insert cares about. + var created = columns.Single(c => c.Value("name") == "created_at"); + Assert.True(created.Value("generated")); + + var amount = columns.Single(c => c.Value("name") == "amount"); + Assert.Equal("decimal", amount.Value("clrType")); + Assert.Equal(10, amount.Value("precision")); + Assert.Equal(2, amount.Value("scale")); + } + + [SkippableFact] + public async Task Discover_finds_procedures_and_functions_with_their_parameters() + { + var adapter = await AdapterAsync(); + + var procedures = await adapter.InvokeAsync("Discover", + new { objectType = "procedure" }, timeoutSeconds: 60); + + var counting = procedures["objects"]!.Single(o => o.Value("name") == "count_orders"); + var parameters = counting["parameters"]!.ToList(); + + Assert.Equal("p_customer", parameters[0].Value("name")); + Assert.Equal("In", parameters[0].Value("direction")); + Assert.Equal("p_total", parameters[1].Value("name")); + Assert.Equal("Out", parameters[1].Value("direction")); + + var functions = await adapter.InvokeAsync("Discover", + new { objectType = "function" }, timeoutSeconds: 60); + + var total = functions["objects"]!.Single(o => o.Value("name") == "total_for"); + + // A function's return type comes from dtd_identifier and is listed once, not twice. + var returns = total["parameters"]!.Where(p => p.Value("direction") == "ReturnValue").ToList(); + Assert.Single(returns); + } + + [SkippableFact] + public async Task Discover_filters_by_name_against_the_database_not_the_page() + { + var adapter = await AdapterAsync(); + + var found = await adapter.InvokeAsync("Discover", + new { objectType = "table", nameLike = "ORDERS" }, timeoutSeconds: 60); + Assert.NotEmpty(found["objects"]!); + + var nothing = await adapter.InvokeAsync("Discover", + new { objectType = "table", nameLike = "no_such_table_anywhere" }, timeoutSeconds: 60); + Assert.Empty(nothing["objects"]!); + } + + // ---------------------------------------------------------------- running + + [SkippableFact] + public async Task Query_returns_rows_for_a_named_statement() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Query", new + { + name = "ordersForCustomer", + parameters = new Dictionary { ["customer"] = "acme" } + }, timeoutSeconds: 60); + + var rows = result["rows"]!.ToList(); + Assert.NotEmpty(rows); + Assert.All(rows, r => Assert.Equal("acme", r.Value("customer"))); + } + + /// + /// The difference that matters: CALL hands back a result set with nothing declared and nothing + /// bound. On PostgreSQL this same shape needs a set-returning function queried with SELECT, and + /// on Oracle an explicit REF CURSOR. + /// + [SkippableFact] + public async Task Call_returns_rows_straight_from_a_procedure() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Call", new + { + name = "ordersByCustomerProc", + parameters = new Dictionary { ["p_customer"] = "globex" } + }, timeoutSeconds: 60); + + var rows = result["rows"]!.ToList(); + Assert.NotEmpty(rows); + Assert.All(rows, r => Assert.Equal("globex", r.Value("customer"))); + } + + [SkippableFact] + public async Task Execute_reports_what_it_changed() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Execute", new + { + name = "insertOrder", + parameters = new Dictionary + { ["id"] = 801, ["customer"] = "inserted", ["amount"] = 12.34 } + }, timeoutSeconds: 60); + + Assert.Equal(1, result.Value("affectedRows")); + + var back = await adapter.InvokeAsync("Query", new + { + name = "ordersForCustomer", + parameters = new Dictionary { ["customer"] = "inserted" } + }, timeoutSeconds: 60); + + Assert.Single(back["rows"]!); + } + + /// + /// A batch is one transaction: either every statement in it happened or none did. Proven by + /// making the second one fail — a duplicate primary key — and then looking for the first. + /// + [SkippableFact] + public async Task A_failed_batch_rolls_back_what_came_before_it() + { + var adapter = await AdapterAsync(); + + await Assert.ThrowsAnyAsync(() => adapter.InvokeAsync("Batch", new + { + statements = new object[] + { + new + { + name = "insertOrder", + parameters = new Dictionary + { ["id"] = 850, ["customer"] = "rolled-back", ["amount"] = 1 } + }, + new + { + // id 1 is seeded, so this violates the primary key. + name = "insertOrder", + parameters = new Dictionary + { ["id"] = 1, ["customer"] = "rolled-back", ["amount"] = 1 } + } + } + }, timeoutSeconds: 60)); + + var back = await adapter.InvokeAsync("Query", new + { + name = "ordersForCustomer", + parameters = new Dictionary { ["customer"] = "rolled-back" } + }, timeoutSeconds: 60); + + Assert.Empty(back["rows"]!); + } + + /// + /// Ad-hoc SQL is refused unless the data source allows it, whatever the engine. The mapper is a + /// template over message content — if it can emit SQL text, every Xchange is an injection + /// vector into the customer's database. + /// + [SkippableFact] + public async Task Sql_sent_with_a_message_is_refused() + { + var adapter = await AdapterAsync(); + + var error = await Assert.ThrowsAnyAsync(() => + adapter.InvokeAsync("Query", + new { sql = $"select * from {MySqlDbFixture.Table}" }, timeoutSeconds: 60)); + + Assert.Contains("does not allow ad-hoc SQL", error.Message); + } + + // ---------------------------------------------------------------- receiving + + [SkippableFact] + public async Task The_receiver_advances_a_cursor_that_survives_a_restart() + { + var adapter = await AdapterAsync(); + var me = Subscription(1); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: me); + var first = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: me); + + Assert.Equal(5, first.Count); + + foreach (var id in first) + { + var file = await adapter.InvokeAsync("GetFile", id, timeoutSeconds: 60, + properties: me); + var data = file.Value("data") ?? file.Value("Data"); + Assert.False(string.IsNullOrWhiteSpace(data)); + + await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 60, properties: me); + } + + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: me); + + Assert.Equal("5", await CursorAsync("receive.cursor.1")); + + var restarted = await Host.RestartAsync(BusAdapters.MySql, _dataSourceId.ToString(), + drain: false); + + await restarted.InvokeAsync("Initialize", timeoutSeconds: 60, properties: me); + var second = await restarted.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: me); + + Assert.Equal(new[] { 6, 7, 8, 9, 10 }, second.Select(KeyOf).OrderBy(i => i)); + } + + [SkippableFact] + public async Task Two_subscriptions_on_one_data_source_do_not_share_a_cursor() + { + var adapter = await AdapterAsync(); + + var first = Subscription(101); + var second = Subscription(202); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: first); + var forFirst = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: first); + foreach (var id in forFirst) + await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 60, properties: first); + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: first); + + Assert.NotEmpty(forFirst); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: second); + var forSecond = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: second); + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: second); + + Assert.Equal( + forFirst.Select(KeyOf).OrderBy(k => k), + forSecond.Select(KeyOf).OrderBy(k => k)); + } + + [SkippableFact] + public async Task Mark_processed_runs_for_each_accepted_row() + { + var adapter = await AdapterAsync(); + var me = Subscription(2); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: me); + var listed = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: me); + + foreach (var id in listed) + await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 60, properties: me); + + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: me); + + var processed = await adapter.InvokeAsync("Query", new { name = "processedOrders" }, + timeoutSeconds: 60); + + Assert.NotEmpty(processed["rows"]!); + } + + // ---------------------------------------------------------------- validating + + [SkippableFact] + public async Task Valid_sql_passes_validation() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = $"select id from {MySqlDbFixture.Table} where id = @id" }, + timeoutSeconds: 60); + + Assert.True(result.Value("ok"), result.Value("error")); + } + + [SkippableFact] + public async Task A_dropped_table_is_caught_before_the_statement_is_stored() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "select 1 from nothing_of_the_sort" }, timeoutSeconds: 60); + + Assert.False(result.Value("ok")); + Assert.False(string.IsNullOrWhiteSpace(result.Value("error"))); + } + + /// + /// The mistake worth naming rather than leaving to a character offset: SQL copied from an + /// Oracle data source, where a parameter is :name, into one where it is @name. + /// + [SkippableFact] + public async Task The_wrong_placeholder_prefix_is_named() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = $"select id from {MySqlDbFixture.Table} where id = :ident" }, + timeoutSeconds: 60); + + Assert.False(result.Value("ok")); + Assert.Contains("@ident", result.Value("error")); + } + + [SkippableFact] + public async Task A_bare_procedure_name_is_accepted_and_says_what_was_not_checked() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "some_procedure" }, timeoutSeconds: 60); + + Assert.True(result.Value("ok")); + Assert.Contains("existence not checked", result.Value("note")); + } + + // ---------------------------------------------------------------- setup + + static int KeyOf(string fileId) => int.Parse(fileId.Substring(fileId.IndexOf(':') + 1)); + + static Dictionary Subscription(int id) => + new() { ["__subscriptionId__"] = id.ToString() }; + + async Task CursorAsync(string name) => + await _fixture.App.Services.GetRequiredService() + .GetAsync(new AdapterStateKey + { + AdapterId = BusAdapters.MySql, + InstanceKey = _dataSourceId.ToString(), + Name = name + }, default); + + async Task CreateDataSourceAsync() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var dataSource = new DataSource + { + Name = $"mysql-{Guid.NewGuid():N}", + AdapterId = BusAdapters.MySql, + Kind = DataSourceKind.Relational, + Properties = Properties(), + SecretProperties = ["Password"] + }; + + db.Add(dataSource); + await db.SaveChangesAsync(); + return dataSource.Id; + } + + Dictionary Properties() => new() + { + ["Host"] = _mysql.Host, + ["Port"] = _mysql.Port.ToString(), + ["Database"] = _mysql.Database, + ["UserName"] = _mysql.User, + ["Password"] = _mysql.Password, + ["MinPoolSize"] = "1", + ["MaxPoolSize"] = "5", + + ["Statements"] = $@"{{ + ""seededOrders"": ""select * from {MySqlDbFixture.Table} where id <= 25 order by id"", + ""ordersForCustomer"": ""select * from {MySqlDbFixture.Table} where customer = @customer order by id"", + ""ordersByCustomerProc"": ""orders_by_customer"", + ""processedOrders"": ""select * from {MySqlDbFixture.Table} where processed = 1 order by id"", + ""insertOrder"": ""insert into {MySqlDbFixture.Table} (id, customer, amount) values (@id, @customer, @amount)"", + + ""earlyOrders"": {{ + ""sql"": ""select * from {MySqlDbFixture.Table} where id > @cursor and id <= 10 order by id"", + ""cursorColumn"": ""id"", + ""keyColumn"": ""id"" + }} + }}", + + ["ReceiveMode"] = "incrementing", + ["ReceiveStatement"] = + $"select * from {MySqlDbFixture.Table} where id > @cursor order by id limit 5", + ["CursorColumn"] = "id", + ["KeyColumn"] = "id", + ["MarkProcessedStatement"] = + $"update {MySqlDbFixture.Table} set processed = 1 where id = @key", + ["ReceiveBatchSize"] = "5" + }; + + async Task StartAdapterAsync() + { + var spec = new AdapterSpec + { + AdapterId = BusAdapters.MySql, + InstanceKey = _dataSourceId.ToString() + }; + + foreach (var kv in Properties()) spec.StartupValues[kv.Key] = kv.Value; + + await Host.StartExclusiveAsync(spec); + } +} diff --git a/SW.Bitween.IntegrationTests/Tests/OracleAdapterTests.cs b/SW.Bitween.IntegrationTests/Tests/OracleAdapterTests.cs index 03ef1a0d..835c606a 100644 --- a/SW.Bitween.IntegrationTests/Tests/OracleAdapterTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/OracleAdapterTests.cs @@ -427,6 +427,69 @@ await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 120, Assert.Equal(new[] { 6, 7, 8, 9, 10 }, keys); } + // ---------------------------------------------------------------- validating + + /// + /// ODP.NET's Prepare is a client-side no-op — Oracle compiles a statement when it is executed, + /// not when it is prepared — so the shared check passed EVERYTHING here, including a select + /// from a table that does not exist. These pin the replacement: DBMS_SQL.PARSE, which compiles + /// and resolves names without running anything. + /// + [SkippableFact] + public async Task Valid_sql_passes_validation() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = $"select id from {OracleFixture.Table} where id = :id" }, + timeoutSeconds: 120); + + Assert.True(result.Value("ok"), result.Value("error")); + } + + [SkippableFact] + public async Task A_dropped_table_is_caught_before_the_statement_is_stored() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "select 1 from nothing_of_the_sort" }, timeoutSeconds: 120); + + Assert.False(result.Value("ok")); + + // ORA-00942: table or view does not exist. Named rather than matched on words, because the + // message is localised and the number is not. + Assert.Contains("ORA-00942", result.Value("error")); + } + + [SkippableFact] + public async Task A_syntax_error_is_caught() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = $"select id from {OracleFixture.Table} where id = @id" }, + timeoutSeconds: 120); + + Assert.False(result.Value("ok")); + + // @ is a database link on Oracle, not a placeholder, so this is ORA-00936 missing + // expression — and the hint says to write :id instead. + Assert.Contains(":id", result.Value("error")); + } + + [SkippableFact] + public async Task A_bare_procedure_name_is_accepted_and_says_what_was_not_checked() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "some_procedure" }, timeoutSeconds: 120); + + Assert.True(result.Value("ok")); + Assert.Contains("existence not checked", result.Value("note")); + } + // ---------------------------------------------------------------- setup async Task CreateDataSourceAsync() diff --git a/SW.Bitween.IntegrationTests/Tests/SqlServerAdapterTests.cs b/SW.Bitween.IntegrationTests/Tests/SqlServerAdapterTests.cs new file mode 100644 index 00000000..62c68fe4 --- /dev/null +++ b/SW.Bitween.IntegrationTests/Tests/SqlServerAdapterTests.cs @@ -0,0 +1,565 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Newtonsoft.Json.Linq; +using SW.Bitween.Domain.DataSources; +using SW.Bitween.IntegrationTests.Fixtures; +using SW.Serverless.Resident; +using Xunit; + +namespace SW.Bitween.IntegrationTests.Tests; + +/// +/// The SQL Server data source provider, against a real SQL Server. +/// +/// The same suite as the others, which is the point: four engines run on one core, so the parts +/// that are shared should behave identically and the parts that differ should differ VISIBLY. +/// SQL Server is the one where the shared statement CHECK could not be shared — its driver refuses +/// to prepare a command whose parameters have no explicit type — so the check has its own tests +/// here rather than relying on the core's. +/// +[Collection("Bitween")] +public class SqlServerAdapterTests : IClassFixture +{ + readonly BitweenFixture _fixture; + readonly SqlServerDbFixture _sqlServer; + + public SqlServerAdapterTests(BitweenFixture fixture, SqlServerDbFixture sqlServer) + { + _fixture = fixture; + _sqlServer = sqlServer; + } + + static readonly SemaphoreSlim Gate = new(1, 1); + static int _dataSourceId; + + IResidentAdapterHost Host => _fixture.App.Services.GetRequiredService(); + + async Task AdapterAsync() + { + Skip.If(_sqlServer.Unavailable != null, $"SQL Server is not available: {_sqlServer.Unavailable}"); + + await Gate.WaitAsync(); + try + { + if (_dataSourceId == 0) + { + _dataSourceId = await CreateDataSourceAsync(); + await StartAdapterAsync(); + } + } + finally + { + Gate.Release(); + } + + return Host.Get(BusAdapters.SqlServer, _dataSourceId.ToString()) + ?? throw new InvalidOperationException("The SQL Server adapter is not running."); + } + + // ---------------------------------------------------------------- configuration + + [SkippableFact] + public async Task Connection_test_reports_each_stage_and_checks_every_statement() + { + var adapter = await AdapterAsync(); + var result = await adapter.InvokeAsync("TestConnection", timeoutSeconds: 60); + + Assert.True(result.Value("ok"), result.ToString()); + + var steps = result["steps"]!.Select(s => s.Value("step")).ToList(); + Assert.Contains("connect", steps); + Assert.Contains("authenticate", steps); + Assert.Contains("privileges", steps); + Assert.Contains("statement:ordersForCustomer", steps); + Assert.All(result["steps"]!, s => Assert.True(s.Value("ok"), s.ToString())); + } + + /// + /// SQL Server is the most capable of the four on paper, and the list says so where it is true — + /// MERGE, OUTPUT and snapshot isolation are all real here — and stays false where the thing + /// exists but is not wired, which is Service Broker's query notifications. + /// + [SkippableFact] + public async Task Describe_reports_the_engine_and_where_it_differs() + { + var adapter = await AdapterAsync(); + var described = await adapter.InvokeAsync("Describe", timeoutSeconds: 60); + + Assert.Equal("SQL Server", described.Value("engine")); + Assert.False(string.IsNullOrWhiteSpace(described.Value("serverVersion"))); + + Assert.True(described.Value("storedProcedures")); + Assert.True(described.Value("procedureResultSets")); + Assert.True(described.Value("merge")); + Assert.True(described.Value("returning")); + + // Table-valued parameters are not an array type: a caller cannot bind a list to one + // parameter the way PostgreSQL allows. + Assert.False(described.Value("arrayTypes")); + + var objects = described["supportedObjects"]!.Select(o => o.Value()).ToList(); + Assert.Contains("sequence", objects); + + var isolation = described["isolationLevels"]!.Select(i => i.Value()).ToList(); + Assert.Contains("Snapshot", isolation); + + // Exists, not wired. Declared false so the UI says "not available" rather than leaving a gap. + Assert.False(described.Value("changeNotification")); + Assert.False(described.Value("logBasedCdc")); + + Assert.NotEmpty(described["privileges"]!); + } + + // ---------------------------------------------------------------- discovery + + [SkippableFact] + public async Task Discover_finds_the_table_with_its_columns_and_key() + { + var adapter = await AdapterAsync(); + var result = await adapter.InvokeAsync("Discover", new + { + objectType = "table", + schema = "sales", + nameLike = SqlServerDbFixture.Table, + includeColumns = true, + includeRowCounts = true + }, timeoutSeconds: 60); + + var table = result["objects"]!.Single(o => o.Value("name") == SqlServerDbFixture.Table); + + Assert.Equal("sales", table.Value("schema")); + + // An extended property, which is where SQL Server keeps what other engines call a comment. + Assert.Equal("Customer orders", table.Value("comment")); + + // The partition stats, not COUNT(*) — so this is an estimate, and asserting an exact + // number would be asserting something the adapter deliberately does not promise. Other + // tests in this class insert rows too, which is the second reason: 25 is the floor. + Assert.True(table.Value("rowCount") >= 25, + $"expected at least the 25 seeded rows, got {table.Value("rowCount")}"); + + var columns = table["columns"]!.ToList(); + Assert.Equal("id", columns[0].Value("name")); + Assert.True(columns[0].Value("primaryKey")); + Assert.False(columns[0].Value("nullable")); + + // max_length is in BYTES and an nvarchar stores two per character, so the declared length + // is half of it — 50, not 100. + var customer = columns.Single(c => c.Value("name") == "customer"); + Assert.Equal("nvarchar(50)", customer.Value("dbType")); + Assert.Equal(50, customer.Value("length")); + + var amount = columns.Single(c => c.Value("name") == "amount"); + Assert.Equal("decimal(10,2)", amount.Value("dbType")); + Assert.Equal("decimal", amount.Value("clrType")); + + // A default constraint is a value the database fills in. + var created = columns.Single(c => c.Value("name") == "created_at"); + Assert.True(created.Value("generated")); + + var processed = columns.Single(c => c.Value("name") == "processed"); + Assert.Equal("bool", processed.Value("clrType")); + } + + /// + /// The schema filter does something, which is why the fixture puts everything in `sales` rather + /// than dbo: against a schema-less seed a filter that was ignored would still pass. + /// + [SkippableFact] + public async Task Discover_filters_by_schema() + { + var adapter = await AdapterAsync(); + + var inSales = await adapter.InvokeAsync("Discover", + new { objectType = "table", schema = "sales" }, timeoutSeconds: 60); + Assert.NotEmpty(inSales["objects"]!); + + var inDbo = await adapter.InvokeAsync("Discover", + new { objectType = "table", schema = "dbo" }, timeoutSeconds: 60); + Assert.Empty(inDbo["objects"]!); + } + + [SkippableFact] + public async Task Discover_finds_procedures_functions_and_sequences() + { + var adapter = await AdapterAsync(); + + var procedures = await adapter.InvokeAsync("Discover", + new { objectType = "procedure", schema = "sales" }, timeoutSeconds: 60); + + var counting = procedures["objects"]!.Single(o => o.Value("name") == "count_orders"); + var parameters = counting["parameters"]!.ToList(); + + // The @ is stripped: a caller binds by name, and the prefix is the driver's business. + Assert.Equal("p_customer", parameters[0].Value("name")); + Assert.Equal("In", parameters[0].Value("direction")); + Assert.Equal("p_total", parameters[1].Value("name")); + Assert.Equal("Out", parameters[1].Value("direction")); + + // An inline table-valued function is a function, not a procedure — the shape a receive + // statement would select from. + var functions = await adapter.InvokeAsync("Discover", + new { objectType = "function", schema = "sales" }, timeoutSeconds: 60); + Assert.Contains(functions["objects"]!, o => o.Value("name") == "orders_for"); + + var sequences = await adapter.InvokeAsync("Discover", + new { objectType = "sequence", schema = "sales" }, timeoutSeconds: 60); + Assert.Contains(sequences["objects"]!, o => o.Value("name") == "bitween_order_seq"); + } + + // ---------------------------------------------------------------- running + + [SkippableFact] + public async Task Query_returns_rows_for_a_named_statement() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Query", new + { + name = "ordersForCustomer", + parameters = new Dictionary { ["customer"] = "acme" } + }, timeoutSeconds: 60); + + var rows = result["rows"]!.ToList(); + Assert.NotEmpty(rows); + Assert.All(rows, r => Assert.Equal("acme", r.Value("customer"))); + } + + [SkippableFact] + public async Task Call_returns_rows_straight_from_a_procedure() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Call", new + { + name = "ordersByCustomerProc", + parameters = new Dictionary { ["p_customer"] = "globex" } + }, timeoutSeconds: 60); + + var rows = result["rows"]!.ToList(); + Assert.NotEmpty(rows); + Assert.All(rows, r => Assert.Equal("globex", r.Value("customer"))); + } + + /// + /// OUTPUT is SQL Server's RETURNING, and the capability list claims it — so a statement using + /// it has to actually work, not merely be declared possible. + /// + [SkippableFact] + public async Task An_insert_with_output_returns_the_row_it_wrote() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Query", new + { + name = "insertOrderOutput", + parameters = new Dictionary + { ["id"] = 902, ["customer"] = "outputted", ["amount"] = 5.5 } + }, timeoutSeconds: 60); + + var rows = result["rows"]!.ToList(); + Assert.Single(rows); + Assert.Equal("outputted", rows[0].Value("customer")); + } + + [SkippableFact] + public async Task Execute_reports_what_it_changed() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("Execute", new + { + name = "insertOrder", + parameters = new Dictionary + { ["id"] = 801, ["customer"] = "inserted", ["amount"] = 12.34 } + }, timeoutSeconds: 60); + + Assert.Equal(1, result.Value("affectedRows")); + } + + [SkippableFact] + public async Task A_failed_batch_rolls_back_what_came_before_it() + { + var adapter = await AdapterAsync(); + + await Assert.ThrowsAnyAsync(() => adapter.InvokeAsync("Batch", new + { + statements = new object[] + { + new + { + name = "insertOrder", + parameters = new Dictionary + { ["id"] = 850, ["customer"] = "rolled-back", ["amount"] = 1 } + }, + new + { + // id 1 is seeded, so this violates the primary key. + name = "insertOrder", + parameters = new Dictionary + { ["id"] = 1, ["customer"] = "rolled-back", ["amount"] = 1 } + } + } + }, timeoutSeconds: 60)); + + var back = await adapter.InvokeAsync("Query", new + { + name = "ordersForCustomer", + parameters = new Dictionary { ["customer"] = "rolled-back" } + }, timeoutSeconds: 60); + + Assert.Empty(back["rows"]!); + } + + [SkippableFact] + public async Task Sql_sent_with_a_message_is_refused() + { + var adapter = await AdapterAsync(); + + var error = await Assert.ThrowsAnyAsync(() => + adapter.InvokeAsync("Query", + new { sql = $"select * from sales.{SqlServerDbFixture.Table}" }, timeoutSeconds: 60)); + + Assert.Contains("does not allow ad-hoc SQL", error.Message); + } + + // ---------------------------------------------------------------- receiving + + [SkippableFact] + public async Task The_receiver_advances_a_cursor_that_survives_a_restart() + { + var adapter = await AdapterAsync(); + var me = Subscription(1); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: me); + var first = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: me); + + Assert.Equal(5, first.Count); + + foreach (var id in first) + { + var file = await adapter.InvokeAsync("GetFile", id, timeoutSeconds: 60, + properties: me); + var data = file.Value("data") ?? file.Value("Data"); + Assert.False(string.IsNullOrWhiteSpace(data)); + + await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 60, properties: me); + } + + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: me); + + Assert.Equal("5", await CursorAsync("receive.cursor.1")); + + var restarted = await Host.RestartAsync(BusAdapters.SqlServer, _dataSourceId.ToString(), + drain: false); + + await restarted.InvokeAsync("Initialize", timeoutSeconds: 60, properties: me); + var second = await restarted.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: me); + + Assert.Equal(new[] { 6, 7, 8, 9, 10 }, second.Select(KeyOf).OrderBy(i => i)); + } + + [SkippableFact] + public async Task Two_subscriptions_on_one_data_source_do_not_share_a_cursor() + { + var adapter = await AdapterAsync(); + + var first = Subscription(101); + var second = Subscription(202); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: first); + var forFirst = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: first); + foreach (var id in forFirst) + await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 60, properties: first); + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: first); + + Assert.NotEmpty(forFirst); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: second); + var forSecond = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: second); + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: second); + + Assert.Equal( + forFirst.Select(KeyOf).OrderBy(k => k), + forSecond.Select(KeyOf).OrderBy(k => k)); + } + + [SkippableFact] + public async Task Mark_processed_runs_for_each_accepted_row() + { + var adapter = await AdapterAsync(); + var me = Subscription(2); + + await adapter.InvokeAsync("Initialize", timeoutSeconds: 60, properties: me); + var listed = await adapter.InvokeAsync>("ListFiles", timeoutSeconds: 60, + properties: me); + + foreach (var id in listed) + await adapter.InvokeAsync("DeleteFile", id, timeoutSeconds: 60, properties: me); + + await adapter.InvokeAsync("Finalize", timeoutSeconds: 60, properties: me); + + var processed = await adapter.InvokeAsync("Query", new { name = "processedOrders" }, + timeoutSeconds: 60); + + Assert.NotEmpty(processed["rows"]!); + } + + // ---------------------------------------------------------------- validating + + /// + /// The check that could not be shared. SqlCommand.Prepare refuses unless every parameter has an + /// explicit type, which a caller checking someone else's SQL does not know — so this adapter + /// asks sp_describe_undeclared_parameters instead, and these prove it answers the same + /// questions the prepared form does everywhere else. + /// + [SkippableFact] + public async Task Valid_sql_passes_validation_even_with_parameters() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = $"select id from sales.{SqlServerDbFixture.Table} where id = @id" }, + timeoutSeconds: 60); + + Assert.True(result.Value("ok"), result.Value("error")); + } + + [SkippableFact] + public async Task A_dropped_table_is_caught_before_the_statement_is_stored() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "select 1 from sales.nothing_of_the_sort" }, timeoutSeconds: 60); + + Assert.False(result.Value("ok")); + Assert.Contains("Invalid object name", result.Value("error")); + } + + [SkippableFact] + public async Task A_syntax_error_is_caught() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "selct 1" }, timeoutSeconds: 60); + + Assert.False(result.Value("ok")); + Assert.False(string.IsNullOrWhiteSpace(result.Value("error"))); + } + + [SkippableFact] + public async Task A_bare_procedure_name_is_accepted_and_says_what_was_not_checked() + { + var adapter = await AdapterAsync(); + + var result = await adapter.InvokeAsync("ValidateStatement", + new { sql = "sales.some_procedure" }, timeoutSeconds: 60); + + Assert.True(result.Value("ok")); + Assert.Contains("existence not checked", result.Value("note")); + } + + // ---------------------------------------------------------------- setup + + static int KeyOf(string fileId) => int.Parse(fileId.Substring(fileId.IndexOf(':') + 1)); + + static Dictionary Subscription(int id) => + new() { ["__subscriptionId__"] = id.ToString() }; + + async Task CursorAsync(string name) => + await _fixture.App.Services.GetRequiredService() + .GetAsync(new AdapterStateKey + { + AdapterId = BusAdapters.SqlServer, + InstanceKey = _dataSourceId.ToString(), + Name = name + }, default); + + async Task CreateDataSourceAsync() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var dataSource = new DataSource + { + Name = $"sqlserver-{Guid.NewGuid():N}", + AdapterId = BusAdapters.SqlServer, + Kind = DataSourceKind.Relational, + Properties = Properties(), + SecretProperties = ["Password"] + }; + + db.Add(dataSource); + await db.SaveChangesAsync(); + return dataSource.Id; + } + + Dictionary Properties() => new() + { + ["Host"] = _sqlServer.Host, + ["Port"] = _sqlServer.Port.ToString(), + ["Database"] = _sqlServer.Database, + ["UserName"] = _sqlServer.User, + ["Password"] = _sqlServer.Password, + ["Schema"] = "sales", + ["MinPoolSize"] = "1", + ["MaxPoolSize"] = "5", + + // The container presents a self-signed certificate, which is the case this setting exists + // for — and the reason the adapter offers it rather than pretending every server has a + // certificate chain that validates. + ["Encrypt"] = "true", + ["TrustServerCertificate"] = "true", + + ["Statements"] = $@"{{ + ""seededOrders"": ""select * from sales.{SqlServerDbFixture.Table} where id <= 25 order by id"", + ""ordersForCustomer"": ""select * from sales.{SqlServerDbFixture.Table} where customer = @customer order by id"", + ""ordersByCustomerProc"": ""sales.orders_by_customer"", + ""ordersForFunction"": ""select * from sales.orders_for(@customer)"", + ""processedOrders"": ""select * from sales.{SqlServerDbFixture.Table} where processed = 1 order by id"", + ""insertOrder"": ""insert into sales.{SqlServerDbFixture.Table} (id, customer, amount) values (@id, @customer, @amount)"", + ""insertOrderOutput"": ""insert into sales.{SqlServerDbFixture.Table} (id, customer, amount) output inserted.* values (@id, @customer, @amount)"", + + ""earlyOrders"": {{ + ""sql"": ""select * from sales.{SqlServerDbFixture.Table} where id > @cursor and id <= 10 order by id"", + ""cursorColumn"": ""id"", + ""keyColumn"": ""id"" + }} + }}", + + ["ReceiveMode"] = "incrementing", + + // TOP rather than LIMIT, and it has to come before the column list — one of the small + // dialect differences a statement writer meets immediately. + ["ReceiveStatement"] = + $"select top 5 * from sales.{SqlServerDbFixture.Table} where id > @cursor order by id", + ["CursorColumn"] = "id", + ["KeyColumn"] = "id", + ["MarkProcessedStatement"] = + $"update sales.{SqlServerDbFixture.Table} set processed = 1 where id = @key", + ["ReceiveBatchSize"] = "5" + }; + + async Task StartAdapterAsync() + { + var spec = new AdapterSpec + { + AdapterId = BusAdapters.SqlServer, + InstanceKey = _dataSourceId.ToString() + }; + + foreach (var kv in Properties()) spec.StartupValues[kv.Key] = kv.Value; + + await Host.StartExclusiveAsync(spec); + } +} diff --git a/SW.Bitween.IntegrationTests/Tests/SqsBusGatewayTests.cs b/SW.Bitween.IntegrationTests/Tests/SqsBusGatewayTests.cs index 710c2b7b..feb0e425 100644 --- a/SW.Bitween.IntegrationTests/Tests/SqsBusGatewayTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/SqsBusGatewayTests.cs @@ -13,6 +13,7 @@ using SW.Bitween.Domain.Gateway; using SW.Bitween.IntegrationTests.Fixtures; using SW.Bitween.Model; +using SW.Bitween.Services.Adapters; using SW.PrimitiveTypes; using SW.Serverless.Resident; using Xunit; @@ -239,6 +240,78 @@ await WaitAsync(async () => await DepthAsync(target) >= 1, TimeSpan.FromSeconds( "the message never arrived on the target queue"); } + /// + /// A DELIVERY sends, which is the half that was missing. Publish had existed since this + /// adapter did and nothing in the pipeline ever called it. + /// + [Fact] + public async Task A_delivery_sends_the_message_it_was_given() + { + var consumed = await fixture.CreateSqsQueueAsync(Unique("handler-in")); + var target = await fixture.CreateSqsQueueAsync(Unique("handler-out")); + + var dataSourceId = await CreateDataSourceAsync(consumed); + await using var adapter = await StartAsync(dataSourceId); + + await adapter.Instance.InvokeAsync("Handle", + new XchangeFile("{\"delivered\":true}", "out.json"), + properties: new Dictionary { ["Endpoint"] = target }); + + await WaitAsync(async () => await DepthAsync(target) >= 1, TimeSpan.FromSeconds(20), + "the delivered message never arrived on the target queue"); + } + + [Fact] + public async Task A_delivery_with_no_endpoint_says_so() + { + var dataSourceId = await CreateDataSourceAsync( + await fixture.CreateSqsQueueAsync(Unique("handler-none"))); + + await using var adapter = await StartAsync(dataSourceId); + + var error = await Assert.ThrowsAnyAsync(() => + adapter.Instance.InvokeAsync("Handle", + new XchangeFile("{}", "out.json"), + properties: new Dictionary())); + + Assert.Contains("no Endpoint", error.Message); + } + + /// + /// The multi-node case. A broker data source is exclusive, but a delivery runs on whichever + /// node picked the message up — so when the owned instance is not here the runtime opens a + /// send-only connection of its own rather than failing. See the RabbitMQ twin of this test + /// for the reasoning in full. + /// + [Fact] + public async Task A_delivery_sends_from_a_node_that_does_not_own_the_connection() + { + var target = await fixture.CreateSqsQueueAsync(Unique("elsewhere-out")); + var dataSourceId = await CreateDataSourceAsync( + await fixture.CreateSqsQueueAsync(Unique("elsewhere-in"))); + + var host = fixture.App.Services.GetRequiredService(); + Assert.Null(host.Get(BusAdapters.Sqs, dataSourceId.ToString())); + + await using var scope = fixture.App.Services.CreateAsyncScope(); + var invoker = scope.ServiceProvider.GetRequiredService(); + + await invoker.InvokeAsync(BusAdapters.Sqs, AdapterRole.Handler, + "Handle", new XchangeFile("{\"fromElsewhere\":true}", "out.json"), + new Dictionary + { + [StartupValuesFiller.DataSourceIdKey] = dataSourceId.ToString(), + ["Endpoint"] = target + }, + Guid.NewGuid().ToString("N")); + + await WaitAsync(async () => await DepthAsync(target) >= 1, TimeSpan.FromSeconds(25), + "a node that does not own the connection could not send"); + + // And it did not take ownership on the way past. + Assert.Null(host.Get(BusAdapters.Sqs, dataSourceId.ToString())); + } + [Fact] public async Task Health_carries_the_queue_detail_from_the_heartbeat() { diff --git a/SW.Bitween.IntegrationTests/Tests/SubscriptionLifecycleTests.cs b/SW.Bitween.IntegrationTests/Tests/SubscriptionLifecycleTests.cs index 8277d44c..8e6139fa 100644 --- a/SW.Bitween.IntegrationTests/Tests/SubscriptionLifecycleTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/SubscriptionLifecycleTests.cs @@ -96,6 +96,48 @@ public async Task An_integration_can_be_created_changed_and_removed() } } + /// + /// Reading a subscription reports which data source it is bound to. + /// + /// Sounds too small to test, and it is exactly the field a merge dropped from the projection + /// once. Nothing failed: the API answered with dataSourceId null while the row held an id, so + /// the UI read every bound subscription as unbound — and saving one from that screen wrote the + /// null back and quietly unbound it. A projection is only as good as the thing that notices a + /// field missing from it. + /// + [Fact] + public async Task Reading_an_integration_reports_the_data_source_it_is_bound_to() + { + var (documentId, partnerId) = await Groundwork(); + var id = await CreateSubscription(Unique("Bound"), documentId, partnerId); + + await using var scope = fixture.CreateScope(); + scope.Superuser(); + var db = scope.ServiceProvider.GetRequiredService(); + + var dataSource = new SW.Bitween.Domain.DataSources.DataSource + { + Name = Unique("bound-source"), + AdapterId = "bitween.db.postgresql", + Kind = SW.Bitween.Domain.DataSources.DataSourceKind.Relational, + Properties = new Dictionary { ["Host"] = "localhost" } + }; + db.Add(dataSource); + await db.SaveChangesAsync(); + + // Set on the row rather than through the update handler, so this tests the READ and + // nothing else. + var stored = await db.Set().SingleAsync(s => s.Id == id); + stored.DataSourceId = dataSource.Id; + await db.SaveChangesAsync(); + + var read = await ActivatorUtilities + .CreateInstance(scope.ServiceProvider) + .Handle(id); + + Assert.Equal(dataSource.Id, ((SubscriptionGet)read).DataSourceId); + } + [Fact] public async Task Deleting_names_the_bus_gateway_route_still_pointing_at_it() { diff --git a/SW.Bitween.UnitTests/CapabilityVersionTests.cs b/SW.Bitween.UnitTests/CapabilityVersionTests.cs new file mode 100644 index 00000000..8d44f8a6 --- /dev/null +++ b/SW.Bitween.UnitTests/CapabilityVersionTests.cs @@ -0,0 +1,48 @@ +using Microsoft.VisualStudio.TestTools.UnitTesting; +using SW.Bitween.Adapters.Db; +using SW.Bitween.Adapters.Db.PostgreSql; + +namespace SW.Bitween.UnitTests; + +/// +/// A capability list is written per engine and is therefore about the engine at its NEWEST. Where +/// a feature arrived in a specific release, the list has to be narrowed once the server version is +/// known — otherwise it promises something an older server will refuse, and the promise is found +/// out on the first message rather than on the screen that made it. +/// +[TestClass] +public class CapabilityVersionTests +{ + [TestMethod] + public void Merge_is_claimed_on_PostgreSQL_15_and_later() + { + Assert.IsTrue(MergeFor("15.6")); + Assert.IsTrue(MergeFor("16.14")); + Assert.IsTrue(MergeFor("17.2")); + } + + [TestMethod] + public void Merge_is_withdrawn_on_PostgreSQL_before_15() + { + // 13 and 14 are both still widely run and were supported into 2025 and 2026. + Assert.IsFalse(MergeFor("13.14")); + Assert.IsFalse(MergeFor("14.11")); + } + + [TestMethod] + public void An_unreadable_version_leaves_the_declared_list_alone() + { + // Better to claim what the engine can do at its newest than to strip a capability because + // a driver reported the version in a shape this did not expect. + Assert.IsTrue(MergeFor(null)); + Assert.IsTrue(MergeFor("")); + Assert.IsTrue(MergeFor("not a version")); + } + + static bool MergeFor(string serverVersion) + { + var described = new DbCapabilities { Merge = true, ServerVersion = serverVersion }; + PostgreSqlDbAdapter.AdjustForVersionForTests(described); + return described.Merge; + } +} diff --git a/SW.Bitween.UnitTests/SW.Bitween.UnitTests.csproj b/SW.Bitween.UnitTests/SW.Bitween.UnitTests.csproj index cfd41424..a4cdf9b7 100644 --- a/SW.Bitween.UnitTests/SW.Bitween.UnitTests.csproj +++ b/SW.Bitween.UnitTests/SW.Bitween.UnitTests.csproj @@ -35,6 +35,9 @@ referencing it from a net10.0 test project is fine and keeps the pure logic — the statement allow-list, cursor formatting — testable without a database. --> + + diff --git a/SW.Bitween.Web/ClientApp/src/api/client.ts b/SW.Bitween.Web/ClientApp/src/api/client.ts index bfff67f1..9daef0e2 100644 --- a/SW.Bitween.Web/ClientApp/src/api/client.ts +++ b/SW.Bitween.Web/ClientApp/src/api/client.ts @@ -72,6 +72,7 @@ import type { TrailEntry, WorkGroupRow, } from "./types"; +import type { SaveResult } from "./http/dataSourceStatements"; /** * The single data-access contract the UI is written against, implemented by @@ -378,7 +379,7 @@ export interface ApiClient { cursorColumn?: string | null; keyColumn?: string | null; }, - ): Promise<{ id: number }>; + ): Promise; updateDataSourceStatement( id: number, changes: { @@ -390,7 +391,7 @@ export interface ApiClient { cursorColumn?: string | null; keyColumn?: string | null; }, - ): Promise; + ): Promise; deleteDataSourceStatement(id: number): Promise; getDataSourceStatementUsage(id: number): Promise; /** The subscription is either an existing id or defined inline; the endpoint commits both as one. */ diff --git a/SW.Bitween.Web/ClientApp/src/api/http/dataSourceStatements.ts b/SW.Bitween.Web/ClientApp/src/api/http/dataSourceStatements.ts index 568e93ca..7cf52b0c 100644 --- a/SW.Bitween.Web/ClientApp/src/api/http/dataSourceStatements.ts +++ b/SW.Bitween.Web/ClientApp/src/api/http/dataSourceStatements.ts @@ -8,6 +8,17 @@ interface SearchyResponse { totalCount: number; } +/** + * What a save answers with. `checked` is false when the database was never asked — the adapter for + * this connection is not running on the node that handled the request, so there was nobody to + * validate against. Worth saying: a save that reported nothing would look exactly like one that + * had been verified. + */ +export interface SaveResult { + id: number; + checked: boolean; +} + interface RawStatement { id: number; dataSourceId: number; @@ -70,10 +81,10 @@ export const dataSourceStatementMethods: Partial = { cursorColumn?: string | null; keyColumn?: string | null; }, - ): Promise<{ id: number }> { + ): Promise { // The data source travels in the body, not the route: POST /datasourcestatements/{id} already // means "update that statement", so a keyed create would collide with it. - const id = await post(`/datasourcestatements`, { + const saved = await post(`/datasourcestatements`, { dataSourceId, name: input.name, sql: input.sql, @@ -83,7 +94,7 @@ export const dataSourceStatementMethods: Partial = { keyColumn: input.keyColumn || null, inactive: false, }); - return { id }; + return saved; }, async updateDataSourceStatement( @@ -97,8 +108,8 @@ export const dataSourceStatementMethods: Partial = { cursorColumn?: string | null; keyColumn?: string | null; }, - ): Promise { - await post(`/datasourcestatements/${id}`, { + ): Promise { + return post(`/datasourcestatements/${id}`, { name: changes.name, sql: changes.sql, description: changes.description ?? null, diff --git a/SW.Bitween.Web/ClientApp/src/pages/data-sources/SchemaBrowser.tsx b/SW.Bitween.Web/ClientApp/src/pages/data-sources/SchemaBrowser.tsx index 0930c34c..c63bb098 100644 --- a/SW.Bitween.Web/ClientApp/src/pages/data-sources/SchemaBrowser.tsx +++ b/SW.Bitween.Web/ClientApp/src/pages/data-sources/SchemaBrowser.tsx @@ -6,6 +6,7 @@ import { Badge, Button, FormError, LoadingBlock } from "../../components/ui/basi import { Select, TextInput } from "../../components/ui/forms"; import { Panel } from "../../components/ui/Panel"; import { + dialectOf, draftNameFor, draftStatementFor, fetchCapabilities, @@ -13,6 +14,7 @@ import { fetchSchemaPage, groupBySchema, type DbObject, + type SqlDialect, } from "./schema"; /** How many objects one page asks for. The adapter clamps anything above 1000. */ @@ -92,6 +94,10 @@ export function SchemaBrowser({ const rows = objects.data?.objects ?? []; const hasMore = objects.data?.hasMore ?? false; + // How to write a placeholder and a row limit for THIS engine. Reported by the adapter rather + // than guessed from the label, so a draft cannot use a prefix the connection will refuse. + const dialect = dialectOf(capabilities.data); + return ( ))} @@ -216,6 +223,7 @@ const LABELS: Record = { function: "Functions", sequence: "Sequences", package: "Packages", + synonym: "Synonyms", }; /** @@ -225,10 +233,12 @@ const LABELS: Record = { function ObjectRow({ dataSourceId, object, + dialect, onUseInStatement, }: { dataSourceId: number; object: DbObject; + dialect: SqlDialect; onUseInStatement?: (draft: { name: string; sql: string; description: string }) => void; }) { const [open, setOpen] = useState(false); @@ -283,7 +293,7 @@ function ObjectRow({ onClick={() => onUseInStatement({ name: draftNameFor(full), - sql: draftStatementFor(full), + sql: draftStatementFor(full, dialect), description: `Generated from ${full.schema}.${full.name}.`, }) } diff --git a/SW.Bitween.Web/ClientApp/src/pages/data-sources/Statements.tsx b/SW.Bitween.Web/ClientApp/src/pages/data-sources/Statements.tsx index 9315d057..604d77bf 100644 --- a/SW.Bitween.Web/ClientApp/src/pages/data-sources/Statements.tsx +++ b/SW.Bitween.Web/ClientApp/src/pages/data-sources/Statements.tsx @@ -9,6 +9,7 @@ import { Badge, Button, FormError, LoadingBlock } from "../../components/ui/basi import { Checkbox, Field, TextInput } from "../../components/ui/forms"; import { ConfirmDialog } from "../../components/ui/overlays"; import { Panel } from "../../components/ui/Panel"; +import { fetchCapabilities } from "./schema"; /** * The SQL this connection is allowed to run. @@ -293,6 +294,16 @@ function StatementForm({ statement?.description ?? seed?.description ?? "", ); const [inactive, setInactive] = useState(statement?.inactive ?? false); + // What this engine spells a placeholder with, reported by the adapter. Named rather than listed + // per engine: there are four now, and a hint that names two of them is wrong for the others. + const capabilities = useQuery({ + queryKey: keys.dataSources.capabilities(dataSourceId), + queryFn: () => fetchCapabilities(dataSourceId), + staleTime: 5 * 60_000, + retry: false, + }); + const prefix = capabilities.data?.parameterPrefix || "@"; + const [cursorColumn, setCursorColumn] = useState(statement?.cursorColumn ?? ""); const [keyColumn, setKeyColumn] = useState(statement?.keyColumn ?? ""); @@ -306,10 +317,16 @@ function StatementForm({ // looks like a button that did nothing. const [justSaved, setJustSaved] = useState(false); + // True when the database never looked at the SQL — the adapter for this connection is not + // running, so there was nobody to ask. Said out loud because a save that reported nothing would + // look exactly like one that had been verified, and the next thing to notice would be a failed + // connection test or a failed message. + const [unchecked, setUnchecked] = useState(false); + const save = useMutation({ mutationFn: async () => { if (statement) { - await api.updateDataSourceStatement(statement.id, { + return api.updateDataSourceStatement(statement.id, { name, sql, description, @@ -318,18 +335,19 @@ function StatementForm({ cursorColumn, keyColumn, }); - } else { - await api.createDataSourceStatement(dataSourceId, { - name, - sql, - description, - cursorColumn, - keyColumn, - }); } + + return api.createDataSourceStatement(dataSourceId, { + name, + sql, + description, + cursorColumn, + keyColumn, + }); }, - onSuccess: () => { + onSuccess: (saved) => { setError(null); + setUnchecked(!saved.checked); // Only for an edit. A create closes the form on success, which says it landed by itself — // and a "Saved" flash on a form that is disappearing is a flicker, not a message. if (statement) { @@ -349,14 +367,17 @@ function StatementForm({
setName(e.target.value)} />