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

using System.Globalization;
using System.Text;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;

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

/// <summary>
///     Folds a set of <see cref="SqlTagDefinition"/>s into the minimum number of parameterized queries
///     (design §3.6) — the driver's batching core. Pure and side-effect free: no connection, no I/O, no clock.
///     <para><b>"Minimum" is bounded by the provider, not only by the group key.</b> A key-value group whose
///     distinct keys exceed <see cref="ISqlDialect.MaxBindParameters"/> is emitted as several plans over the
///     same group key rather than as one statement the backend refuses — see <see cref="AddKeyValuePlans"/>.</para>
///     <para><b>The GroupKey is the correctness contract.</b> Two tags share a query <em>only</em> when every
///     field that shapes the result set is identical. Too coarse and a tag silently reads a neighbour's cell;
///     too fine and batching degenerates to one round-trip per tag. Both directions are pinned by
///     <c>SqlGroupPlannerTests</c>.</para>
///     <para><b>Injection boundary (design §8.1).</b> Identifiers reach the command text through
///     <see cref="ISqlDialect.QuoteIdentifier"/> and nothing else; every authored value becomes a
///     <c>k0</c>/<c>w</c> parameter — spelled with the dialect's own <see cref="ISqlDialect.ParameterMarker"/>
///     (<c>@k0</c> on SQL Server, <c>:p_k0</c> on Oracle) — with the value carried in
///     <see cref="SqlQueryPlan.Parameters"/>. There is no interpolation of tag content into SQL anywhere in
///     this type.</para>
///     <para><b>The base names are this planner's, the spelling is the dialect's.</b> Each plan carries the
///     markers it emitted <em>and</em> the matching <see cref="ISqlDialect.ParameterBindName"/>s, so the
///     reader binds what the planner minted instead of re-deriving a provider convention — the two differ on
///     ODP.NET (<c>:p_k0</c> in the text, bare <c>p_k0</c> on the parameter).</para>
/// </summary>
public static class SqlGroupPlanner
{
    /// <summary>The base-name prefix for a key-value model's <c>IN</c>-list parameters (<c>k0…kN</c>).</summary>
    private const string KeyBaseNamePrefix = "k";

    /// <summary>The single parameter base name for a wide-row model's row selector.</summary>
    private const string RowSelectorBaseName = "w";

    /// <summary>
    ///     Groups <paramref name="tags"/> by source and emits one <see cref="SqlQueryPlan"/> per group — or
    ///     <b>several</b>, when a key-value group holds more distinct keys than the dialect's
    ///     <see cref="ISqlDialect.MaxBindParameters"/> allows in one command (see
    ///     <see cref="AddKeyValuePlans"/>). Plans from one group share that group's key.
    ///     <para>Deterministic: groups appear in the input order of their first member, a split group's chunks
    ///     stay adjacent and in key order, members keep their input order within a chunk, and parameters keep
    ///     the first-appearance order of their distinct values — so a given tag list always yields
    ///     byte-identical SQL.</para>
    /// </summary>
    /// <param name="tags">The tags to plan. An empty sequence yields no plans.</param>
    /// <param name="dialect">
    ///     Supplies identifier quoting, the provider's row-limit syntax, and its per-command bind-parameter
    ///     ceiling.
    /// </param>
    /// <returns>At least one plan per distinct group key; more when a key-value group had to be chunked.</returns>
    /// <exception cref="ArgumentNullException"><paramref name="tags"/> or <paramref name="dialect"/> is null.</exception>
    /// <exception cref="ArgumentException">
    ///     A definition is missing a field its model requires (e.g. a wide-row tag with no row selector), or an
    ///     identifier cannot be quoted (an empty table-name part, a control character — see
    ///     <see cref="ISqlDialect.QuoteIdentifier"/>). <see cref="SqlEquipmentTagParser"/> rejects all of these
    ///     up front, so reaching one here means a caller built a definition by hand with a broken invariant.
    /// </exception>
    /// <exception cref="NotSupportedException">A tag carries a model this planner does not know.</exception>
    /// <param name="queries">
    ///     The driver's validated named queries, which a <see cref="SqlTagModel.Query"/> tag needs: every tag
    ///     naming one query folds into ONE plan running that query's own statement (design §5.4). Null when the
    ///     driver declares none; a Query tag naming a query this does not hold throws
    ///     <see cref="ArgumentException"/> (the driver drops such tags at Initialize, so that is a broken caller
    ///     invariant).
    /// </param>
    public static IReadOnlyList<SqlQueryPlan> Plan(
        IEnumerable<SqlTagDefinition> tags,
        ISqlDialect dialect,
        IReadOnlyDictionary<string, SqlNamedQuery>? queries = null)
    {
        ArgumentNullException.ThrowIfNull(tags);
        ArgumentNullException.ThrowIfNull(dialect);

        // Ordered grouping: the dictionary decides membership, the list decides emission order.
        var order = new List<object>();
        var groups = new Dictionary<object, List<SqlTagDefinition>>();

        foreach (var tag in tags)
        {
            ArgumentNullException.ThrowIfNull(tag);
            var key = BuildGroupKey(tag);
            if (!groups.TryGetValue(key, out var members))
            {
                members = [];
                groups.Add(key, members);
                order.Add(key);
            }

            members.Add(tag);
        }

        var plans = new List<SqlQueryPlan>(order.Count);
        foreach (var key in order)
        {
            var members = groups[key];
            switch (members[0].Model)
            {
                case SqlTagModel.KeyValue:
                    // One group can yield SEVERAL plans: the IN list is chunked at the dialect's ceiling.
                    AddKeyValuePlans(plans, key, members, dialect);
                    break;
                case SqlTagModel.WideRow:
                    plans.Add(BuildWideRowPlan(key, members, dialect));
                    break;
                case SqlTagModel.Query:
                    plans.Add(BuildQueryPlan(key, members, dialect, queries));
                    break;
                default:
                    throw UnsupportedModel(members[0]);
            }
        }

        return plans;
    }

    /// <summary>
    ///     Composes the equality key that decides which tags share a query. Every model produces a 5-tuple whose
    ///     first element is the model, so keys of different models can never compare equal.
    ///     <list type="bullet">
    ///         <item>
    ///             <see cref="SqlTagModel.KeyValue"/> → <c>(model, table, keyColumn, valueColumn, timestampColumn)</c>.
    ///             Every one of those shapes the emitted SELECT: sharing a plan across two <c>valueColumn</c>s
    ///             would make one tag publish the other's column.
    ///         </item>
    ///         <item>
    ///             <see cref="SqlTagModel.WideRow"/> → <c>(model, table, whereColumn, whereValue, topByTimestamp)</c>
    ///             — i.e. the table plus the whole row selector, exactly design §5.3's
    ///             <c>(table, rowSelector)</c>. The members' own <c>columnName</c>/<c>timestampColumn</c> are
    ///             deliberately <b>not</b> in the key: differing there is the point of the model (many columns,
    ///             one row), and they are added to the SELECT list instead.
    ///         </item>
    ///     </list>
    ///     <para>String comparison is <b>ordinal</b>. SQL Server object names are usually case-insensitive, so
    ///     two tags authored <c>dbo.T</c> and <c>DBO.T</c> plan as two groups. That costs one extra round-trip
    ///     and is never wrong; guessing the server's collation here could fold two genuinely different sources.</para>
    /// </summary>
    private static object BuildGroupKey(SqlTagDefinition tag) => tag.Model switch
    {
        SqlTagModel.KeyValue => (tag.Model, tag.Table, tag.KeyColumn, tag.ValueColumn, tag.TimestampColumn),
        SqlTagModel.WideRow => (tag.Model, tag.Table, tag.RowSelectorColumn, tag.RowSelectorValue,
            tag.RowSelectorTopByTimestamp),
        // One execution per named query, whatever columns or rows its members pick out of the result.
        SqlTagModel.Query => (tag.Model, tag.QueryName, (string?)null, (string?)null, (string?)null),
        _ => throw UnsupportedModel(tag),
    };

    /// <summary>
    ///     Emits one plan for a named query: the query's own (already-validated, read-only) statement, with
    ///     each declared parameter spelled by the dialect's marker and bound under its bind name.
    ///     <para><b>No identifier from a tag reaches this text.</b> A Query tag's <c>column</c>,
    ///     <c>rowKey.column</c> and <c>timestampColumn</c> are matched against the result set's column names
    ///     by the reader; they are never quoted into SQL, so they need no catalog gate.</para>
    ///     <para><b>The row cap is client-side only.</b> An arbitrary statement cannot be wrapped in a
    ///     provider row limit portably (a T-SQL CTE cannot sit in a derived table), so
    ///     <see cref="SqlQueryPlan.RowCap"/> is the query's <c>maxRows</c> and the reader stops reading one row
    ///     past it.</para>
    /// </summary>
    private static SqlQueryPlan BuildQueryPlan(
        object groupKey,
        List<SqlTagDefinition> members,
        ISqlDialect dialect,
        IReadOnlyDictionary<string, SqlNamedQuery>? queries)
    {
        var first = members[0];
        var name = first.QueryName;
        if (string.IsNullOrWhiteSpace(name)) throw MissingField(nameof(SqlTagDefinition.QueryName), first);
        if (queries is null || !queries.TryGetValue(name, out var query))
            throw new ArgumentException(
                $"Sql tag '{first.Name}' reads named query '{name}', which the driver config does not define.",
                nameof(queries));

        var markers = new List<string>(query.ParameterNames.Count);
        var bindNames = new List<string>(query.ParameterNames.Count);
        foreach (var parameter in query.ParameterNames)
        {
            markers.Add(dialect.ParameterMarker(parameter));
            bindNames.Add(dialect.ParameterBindName(parameter));
        }

        return new SqlQueryPlan(
            SqlTagModel.Query, groupKey, query.Sql, markers, bindNames, query.ParameterValues, members,
            SelectedColumns: [], RowCap: query.MaxRows);
    }

    /// <summary>
    ///     Emits one key-value plan per <b>chunk</b> of the group's distinct keys, sized by
    ///     <see cref="ISqlDialect.MaxBindParameters"/>. A group at or under the ceiling — every ordinary group
    ///     — still yields exactly one plan.
    ///     <para><b>Why chunk rather than let the provider refuse.</b> Every backend caps parameters per
    ///     command (2100 on SQL Server, 1000 <c>IN</c>-list entries on Oracle), and a key-value (EAV) staging
    ///     table is precisely the shape that attracts thousands of tags on <em>one</em> table — which the
    ///     group-key rule folds into one group. Over the cap the provider rejects the statement <em>after</em>
    ///     the connection opened, so every tag in the group publishes <c>BadCommunicationError</c> on every
    ///     poll, permanently, and the driver reports a communication fault against a database that is
    ///     answering.</para>
    ///     <para><b>Chunks share the group key, and no consumer needs to change.</b> The reader executes one
    ///     command per plan and slices each plan's own <see cref="SqlQueryPlan.Members"/> back, so several
    ///     plans over one group key cost extra round-trips and nothing else. The alternative — a deploy-time
    ///     tag-count gate — would refuse a configuration the driver can serve perfectly well.</para>
    ///     <para><b>Members follow their key, never their position.</b> Two tags authored on the same
    ///     <c>keyValue</c> land in the same chunk (one bound parameter, both fed); a member placed in a chunk
    ///     that does not bind its key would read a result set that cannot contain its row and publish
    ///     <c>BadNoData</c> forever.</para>
    /// </summary>
    private static void AddKeyValuePlans(
        List<SqlQueryPlan> plans, object groupKey, List<SqlTagDefinition> members, ISqlDialect dialect)
    {
        var ceiling = dialect.MaxBindParameters;
        if (ceiling < 1)
        {
            // Not a silent clamp: chunking to zero would loop forever, and a one-per-chunk fallback would
            // hide a broken dialect behind a poll that suddenly costs one round-trip per tag.
            throw new ArgumentException(
                $"The {dialect.Provider} dialect reports a per-command bind-parameter ceiling of {ceiling}; " +
                "it must be at least 1.", nameof(dialect));
        }

        // Distinct keys in first-appearance order decide the chunk boundaries; every member then follows its
        // own key. Both halves keep the planner's determinism: a given tag list yields byte-identical SQL.
        var chunkOfKey = new Dictionary<string, int>(StringComparer.Ordinal);
        foreach (var member in members)
        {
            var keyValue = member.KeyValue
                           ?? throw MissingField(nameof(SqlTagDefinition.KeyValue), member);
            if (!chunkOfKey.ContainsKey(keyValue)) chunkOfKey[keyValue] = chunkOfKey.Count / ceiling;
        }

        var chunkCount = ((chunkOfKey.Count - 1) / ceiling) + 1;
        if (chunkCount == 1)
        {
            plans.Add(BuildKeyValuePlan(groupKey, members, dialect));
            return;
        }

        var chunks = new List<SqlTagDefinition>[chunkCount];
        for (var c = 0; c < chunkCount; c++) chunks[c] = [];
        foreach (var member in members) chunks[chunkOfKey[member.KeyValue!]].Add(member);

        foreach (var chunk in chunks) plans.Add(BuildKeyValuePlan(groupKey, chunk, dialect));
    }

    /// <summary>
    ///     Emits <c>SELECT &lt;key&gt;, &lt;value&gt;[, &lt;ts&gt;] FROM &lt;table&gt; WHERE &lt;key&gt; IN (@k0..@kN)</c>
    ///     for one chunk of a group. The key column is selected because the reader indexes rows by it to slice
    ///     values back per member.
    ///     <para>Keys are de-duplicated (ordinal) before binding: two tags authored against the same
    ///     <c>keyValue</c> bind one parameter but stay two members, so both nodes are fed from the one row.</para>
    ///     <para>Markers are numbered from <c>k0</c> within the chunk — each chunk is its own command, so
    ///     carrying a group-wide index forward would leave a three-parameter command spelling
    ///     <c>IN (@k3, @k4, @k5)</c>.</para>
    /// </summary>
    private static SqlQueryPlan BuildKeyValuePlan(
        object groupKey, List<SqlTagDefinition> members, ISqlDialect dialect)
    {
        var first = members[0];
        var keyColumn = RequireIdentifier(first.KeyColumn, nameof(SqlTagDefinition.KeyColumn), first);
        var valueColumn = RequireIdentifier(first.ValueColumn, nameof(SqlTagDefinition.ValueColumn), first);
        var timestampColumn = Blank(first.TimestampColumn) ? null : first.TimestampColumn;

        var selected = new List<string>(3);
        AddDistinct(selected, keyColumn);
        AddDistinct(selected, valueColumn);
        if (timestampColumn is not null) AddDistinct(selected, timestampColumn);

        // A keyValue may legitimately be the empty string, so only null is a broken invariant.
        var markers = new List<string>(members.Count);
        var bindNames = new List<string>(members.Count);
        var values = new List<object?>(members.Count);
        var seen = new HashSet<string>(StringComparer.Ordinal);
        foreach (var member in members)
        {
            var keyValue = member.KeyValue
                           ?? throw MissingField(nameof(SqlTagDefinition.KeyValue), member);
            if (!seen.Add(keyValue)) continue;
            var baseName = KeyBaseNamePrefix + values.Count.ToString(CultureInfo.InvariantCulture);
            markers.Add(dialect.ParameterMarker(baseName));
            bindNames.Add(dialect.ParameterBindName(baseName));
            values.Add(keyValue);
        }

        // Row cap (R06:S-1): one row per distinct key is all this plan can use. It reads up to TWICE that, so a
        // handful of stray duplicates stays inside the cap and is attributed PER KEY by the reader (only the
        // duplicated key's tags go BadConfigurationError) — and asks for one row more than that, so only a
        // gross excess (a history/events table, unattributable by construction because the rest is unread)
        // fails the whole group. Memory stays O(keys) either way.
        var rowCap = values.Count * 2;
        var sql = new StringBuilder("SELECT ")
            .Append(dialect.RowLimitPrefix(rowCap + 1))
            .Append(QuoteList(selected, dialect))
            .Append(" FROM ").Append(QuoteQualifiedName(first.Table, dialect))
            .Append(" WHERE ").Append(dialect.QuoteIdentifier(keyColumn))
            .Append(" IN (").AppendJoin(", ", markers).Append(')')
            .Append(dialect.RowLimitSuffix(rowCap + 1))
            .ToString();

        return new SqlQueryPlan(
            SqlTagModel.KeyValue, groupKey, sql, markers, bindNames, values, members, selected, rowCap);
    }

    /// <summary>
    ///     Emits <c>SELECT &lt;cols&gt; FROM &lt;table&gt; WHERE &lt;whereColumn&gt; = @w</c>, or — for a
    ///     <c>topByTimestamp</c> selector — the single-row form
    ///     <c>SELECT &lt;limit&gt; &lt;cols&gt; FROM &lt;table&gt; ORDER BY &lt;ts&gt; DESC &lt;limit&gt;</c>, whose
    ///     row limit comes from <see cref="ISqlDialect.SingleRowLimitPrefix"/> /
    ///     <see cref="ISqlDialect.SingleRowLimitSuffix"/> (T-SQL fills the prefix: <c>SELECT TOP 1 …</c>).
    ///     <para>The SELECT list is each member's <c>columnName</c> (distinct, in first-appearance order),
    ///     followed by each distinct member <c>timestampColumn</c> — so members of one row may carry different
    ///     timestamp columns without splitting the group. The where-column itself is <b>not</b> selected: every
    ///     row in the result already matches the bound value, so reading it back adds nothing.</para>
    /// </summary>
    private static SqlQueryPlan BuildWideRowPlan(
        object groupKey, List<SqlTagDefinition> members, ISqlDialect dialect)
    {
        var first = members[0];

        var selected = new List<string>(members.Count + 1);
        foreach (var member in members)
            AddDistinct(selected, RequireIdentifier(member.ColumnName, nameof(SqlTagDefinition.ColumnName), member));
        foreach (var member in members)
            if (!Blank(member.TimestampColumn))
                AddDistinct(selected, member.TimestampColumn!);

        var table = QuoteQualifiedName(first.Table, dialect);
        var columns = QuoteList(selected, dialect);

        // A where-pair wins over topByTimestamp; the parser already guarantees exactly one is populated.
        if (!Blank(first.RowSelectorColumn) && first.RowSelectorValue is not null)
        {
            // Capped at TWO rows (R06:S-1): the selector must identify one row, and the second is the proof it
            // does not. A TOP 1 alone would hide that ambiguity (see SqlPollReader.SelectWideRow); no cap at all
            // materialised every matching row of an events table on every poll.
            var marker = dialect.ParameterMarker(RowSelectorBaseName);
            var sql = string.Concat(
                "SELECT ", dialect.RowLimitPrefix(2), columns, " FROM ", table,
                " WHERE ", dialect.QuoteIdentifier(first.RowSelectorColumn!), " = ", marker,
                dialect.RowLimitSuffix(2));
            return new SqlQueryPlan(
                SqlTagModel.WideRow, groupKey, sql,
                [marker], [dialect.ParameterBindName(RowSelectorBaseName)],
                [first.RowSelectorValue], members, selected, RowCap: 1);
        }

        if (!Blank(first.RowSelectorTopByTimestamp))
        {
            // The row limit is dialect syntax and sits at opposite ends of the statement per provider
            // (T-SQL "SELECT TOP 1 …", SQLite/Postgres/MySQL "… LIMIT 1"), so both ends are emitted
            // unconditionally and the dialect decides which one it fills. See ISqlDialect.SingleRowLimitPrefix.
            var sql = string.Concat(
                "SELECT ", dialect.SingleRowLimitPrefix, columns, " FROM ", table,
                " ORDER BY ", dialect.QuoteIdentifier(first.RowSelectorTopByTimestamp!), " DESC",
                dialect.SingleRowLimitSuffix);
            return new SqlQueryPlan(SqlTagModel.WideRow, groupKey, sql, [], [], [], members, selected);
        }

        throw new ArgumentException(
            $"Wide-row tag '{first.Name}' has no row selector: author either a rowSelector.whereColumn/" +
            "whereValue pair or a rowSelector.topByTimestamp column.", nameof(members));
    }

    /// <summary>
    ///     Quotes a possibly multi-part object name by splitting on <c>.</c> and quoting each part —
    ///     <c>dbo.TagValues</c> becomes <c>[dbo].[TagValues]</c>.
    ///     <para><see cref="ISqlDialect.QuoteIdentifier"/> quotes exactly <b>one</b> part; handing it the dotted
    ///     form would yield <c>[dbo.TagValues]</c>, which is safe but names an object that does not exist. An
    ///     empty part (<c>dbo..T</c>) is rejected by the dialect rather than emitted. As a consequence an object
    ///     whose real name contains a literal <c>.</c> cannot be authored — an accepted v1 limitation.</para>
    /// </summary>
    private static string QuoteQualifiedName(string name, ISqlDialect dialect)
    {
        if (Blank(name))
            throw new ArgumentException("A Sql tag's 'table' must be a non-empty object name.", nameof(name));

        return string.Join('.', name.Split('.').Select(dialect.QuoteIdentifier));
    }

    /// <summary>Quotes each already-de-duplicated column name and joins them into a SELECT list.</summary>
    private static string QuoteList(IReadOnlyList<string> columns, ISqlDialect dialect)
        => string.Join(", ", columns.Select(dialect.QuoteIdentifier));

    /// <summary>Appends <paramref name="value"/> unless an ordinal-equal entry is already present.</summary>
    private static void AddDistinct(List<string> target, string value)
    {
        if (!target.Contains(value, StringComparer.Ordinal)) target.Add(value);
    }

    /// <summary>True when the string is null, empty, or all whitespace — i.e. cannot be an identifier.</summary>
    private static bool Blank(string? value) => string.IsNullOrWhiteSpace(value);

    /// <summary>Returns a required identifier field, or throws naming both the tag and the field.</summary>
    private static string RequireIdentifier(string? value, string field, SqlTagDefinition tag)
        => Blank(value) ? throw MissingField(field, tag) : value!;

    private static ArgumentException MissingField(string field, SqlTagDefinition tag)
        => new($"Sql tag '{tag.Name}' ({tag.Model}) is missing the required '{field}'.");

    private static NotSupportedException UnsupportedModel(SqlTagDefinition tag)
        => new($"Sql tag '{tag.Name}' uses the {tag.Model} model, which the poll planner does not support.");
}