lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogLoader.cs

using System.Data;
using System.Data.Common;

namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;

/// <summary>
///     Loads the <see cref="SqlCatalog"/> slice the authored tags need, over one connection, using only the
///     dialect's catalog SQL (design §8.1).
/// </summary>
/// <remarks>
///     <para><b>Bounded by what is authored, not by what exists.</b> One query for the schema list, one for
///     the default schema, then one <see cref="ISqlDialect.ListTablesSql"/> per distinct authored schema and
///     one <see cref="ISqlDialect.ListColumnsSql"/> per distinct authored relation. A driver polling three
///     tables issues a handful of round-trips at Initialize and none thereafter; it never enumerates a
///     warehouse.</para>
///     <para><b>Every parameter is bound.</b> Authored names reach the catalog queries as
///     <c>@schema</c>/<c>@table</c> parameters — the same discipline the schema browser follows — so loading
///     the allow-list cannot itself be an injection vector. No identifier is quoted into text on this path.</para>
///     <para><b>Failures throw; they do not degrade to an empty catalog.</b> An empty catalog would reject
///     every tag, which is indistinguishable at the OPC UA surface from a database whose objects were all
///     dropped. A caller that cannot read the catalog has not learned that the tags are invalid — it has
///     learned nothing — and must fail its Initialize so the driver retries rather than serving a
///     confidently-empty address space.</para>
/// </remarks>
internal static class SqlCatalogLoader
{
    /// <summary>Column alias <see cref="ISqlDialect.ListSchemasSql"/> projects.</summary>
    private const string SchemaColumn = "TABLE_SCHEMA";

    /// <summary>Column alias <see cref="ISqlDialect.ListTablesSql"/> projects.</summary>
    private const string TableNameColumn = "TABLE_NAME";

    /// <summary>Column alias <see cref="ISqlDialect.ListColumnsSql"/> projects.</summary>
    private const string ColumnNameColumn = "COLUMN_NAME";

    /// <summary>
    ///     Reads the catalog slice covering <paramref name="authoredTables"/>.
    /// </summary>
    /// <param name="connection">An <b>already-open</b> connection. Not disposed here; the caller owns it.</param>
    /// <param name="dialect">Supplies the catalog SQL.</param>
    /// <param name="authoredTables">The distinct authored <c>table</c> strings, as written in the blobs.</param>
    /// <param name="commandTimeout">Per-command server-side backstop.</param>
    /// <param name="cancellationToken">Cancellation token for the operation.</param>
    /// <returns>The loaded catalog.</returns>
    /// <exception cref="InvalidOperationException">The connection can see no schemas at all.</exception>
    internal static async Task<SqlCatalog> LoadAsync(
        DbConnection connection,
        ISqlDialect dialect,
        IReadOnlyCollection<string> authoredTables,
        TimeSpan commandTimeout,
        CancellationToken cancellationToken)
    {
        var timeoutSeconds = Math.Max(1, (int)Math.Ceiling(commandTimeout.TotalSeconds));

        var schemas = await QueryColumnAsync(
            connection, dialect.ListSchemasSql, SchemaColumn,
            static _ => { }, timeoutSeconds, cancellationToken).ConfigureAwait(false);

        // Zero schemas is a permissions or visibility symptom, never a real database. Rejecting every tag on
        // that basis would report an authoring fault for what is actually a grant problem, and would send the
        // operator to the wrong system — the same misdiagnosis the driver's health classification avoids.
        if (schemas.Count == 0)
            throw new InvalidOperationException(
                "the catalog returned no schemas at all; the connecting principal most likely cannot read " +
                "the catalog views, which is a grant problem rather than a tag-authoring one");

        var defaultSchema = await ReadDefaultSchemaAsync(
            connection, dialect, timeoutSeconds, cancellationToken).ConfigureAwait(false);

        var tablesBySchema = new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal);
        var columnsByTable = new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal);

        // Resolving here goes through SqlCatalog.TryMatch — the SAME matcher the gate will use — so the
        // loader and the gate can never disagree about which relation an authored name resolved to. Loading
        // one table and validating against another would be a silent wrong-column defect.
        foreach (var authored in authoredTables)
        {
            if (string.IsNullOrWhiteSpace(authored)) continue;

            var parts = authored.Split('.');
            if (parts.Length > 2) continue;   // the gate rejects these by name; nothing to load.

            var authoredTable = parts[^1];
            string schema;
            if (parts.Length == 2)
            {
                if (!SqlCatalog.TryMatch(schemas, parts[0], out schema)) continue;
            }
            else
            {
                schema = defaultSchema;
            }

            if (!tablesBySchema.TryGetValue(schema, out var tables))
            {
                tables = await QueryColumnAsync(
                    connection, dialect.ListTablesSql, TableNameColumn,
                    command => Bind(command, "@schema", schema),
                    timeoutSeconds, cancellationToken).ConfigureAwait(false);
                tablesBySchema[schema] = tables;
            }

            if (!SqlCatalog.TryMatch(tables, authoredTable, out var table)) continue;

            var key = SqlCatalog.QualifiedKey(schema, table);
            if (columnsByTable.ContainsKey(key)) continue;

            columnsByTable[key] = await QueryColumnAsync(
                connection, dialect.ListColumnsSql, ColumnNameColumn,
                command =>
                {
                    Bind(command, "@schema", schema);
                    Bind(command, "@table", table);
                },
                timeoutSeconds, cancellationToken).ConfigureAwait(false);
        }

        return new SqlCatalog(defaultSchema, schemas, tablesBySchema, columnsByTable);
    }

    /// <summary>
    ///     Reads the connection's default schema, falling back to the sole visible schema when the dialect's
    ///     query answers null/blank.
    /// </summary>
    /// <remarks>
    ///     A null answer is possible (a login with no default schema mapped). Falling back to the single
    ///     visible schema keeps the common one-schema database working; with several visible schemas there is
    ///     no defensible guess, so unqualified names simply will not resolve and the gate says so by name.
    /// </remarks>
    private static async Task<string> ReadDefaultSchemaAsync(
        DbConnection connection, ISqlDialect dialect, int timeoutSeconds, CancellationToken cancellationToken)
    {
        var command = connection.CreateCommand();
        await using (command.ConfigureAwait(false))
        {
            command.CommandText = dialect.DefaultSchemaSql;
            command.CommandTimeout = timeoutSeconds;
            var value = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
            var schema = value is null or DBNull ? null : value.ToString();
            return string.IsNullOrWhiteSpace(schema) ? string.Empty : schema;
        }
    }

    /// <summary>Runs one catalog query and projects a single string column from every row.</summary>
    private static async Task<IReadOnlyList<string>> QueryColumnAsync(
        DbConnection connection,
        string sql,
        string columnName,
        Action<DbCommand> bind,
        int timeoutSeconds,
        CancellationToken cancellationToken)
    {
        var results = new List<string>();

        var command = connection.CreateCommand();
        await using (command.ConfigureAwait(false))
        {
            command.CommandText = sql;
            command.CommandTimeout = timeoutSeconds;
            bind(command);

            var reader = await command
                .ExecuteReaderAsync(CommandBehavior.SingleResult, cancellationToken).ConfigureAwait(false);
            await using (reader.ConfigureAwait(false))
            {
                var ordinal = -1;
                while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
                {
                    if (ordinal < 0) ordinal = reader.GetOrdinal(columnName);
                    if (reader.IsDBNull(ordinal)) continue;
                    var value = reader.GetValue(ordinal)?.ToString();
                    if (!string.IsNullOrWhiteSpace(value)) results.Add(value);
                }
            }
        }

        return results;
    }

    /// <summary>Binds one catalog-query parameter — the only way an authored name reaches the database here.</summary>
    private static void Bind(DbCommand command, string name, string value)
    {
        var parameter = command.CreateParameter();
        parameter.ParameterName = name;
        parameter.DbType = DbType.String;
        parameter.Value = value;
        command.Parameters.Add(parameter);
    }
}